You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2015/04/24 22:32:40 UTC
[2/4] storm git commit: STORM-583: initial import of donated code
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
new file mode 100755
index 0000000..5600873
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
@@ -0,0 +1,150 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import org.apache.qpid.amqp_1_0.client.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import backtype.storm.metric.api.CountMetric;
+import backtype.storm.metric.api.MeanReducer;
+import backtype.storm.metric.api.ReducedMetric;
+
+import org.apache.storm.eventhubs.client.Constants;
+import org.apache.storm.eventhubs.client.EventHubClient;
+import org.apache.storm.eventhubs.client.EventHubException;
+import org.apache.storm.eventhubs.client.EventHubReceiver;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.qpid.amqp_1_0.type.Section;
+import org.apache.qpid.amqp_1_0.type.Symbol;
+import org.apache.qpid.amqp_1_0.type.messaging.MessageAnnotations;
+
+public class EventHubReceiverImpl implements IEventHubReceiver {
+ private static final Logger logger = LoggerFactory.getLogger(EventHubReceiverImpl.class);
+ private static final Symbol OffsetKey = Symbol.valueOf("x-opt-offset");
+ private static final Symbol SequenceNumberKey = Symbol.valueOf("x-opt-sequence-number");
+
+ private final String connectionString;
+ private final String entityName;
+ private final String partitionId;
+ private final int defaultCredits;
+
+ private EventHubReceiver receiver;
+ private String lastOffset = null;
+ private ReducedMetric receiveApiLatencyMean;
+ private CountMetric receiveApiCallCount;
+ private CountMetric receiveMessageCount;
+
+ public EventHubReceiverImpl(EventHubSpoutConfig config, String partitionId) {
+ this.connectionString = config.getConnectionString();
+ this.entityName = config.getEntityPath();
+ this.defaultCredits = config.getReceiverCredits();
+ this.partitionId = partitionId;
+ receiveApiLatencyMean = new ReducedMetric(new MeanReducer());
+ receiveApiCallCount = new CountMetric();
+ receiveMessageCount = new CountMetric();
+ }
+
+ @Override
+ public void open(IEventHubReceiverFilter filter) throws EventHubException {
+ logger.info("creating eventhub receiver: partitionId=" + partitionId + ", offset=" + filter.getOffset()
+ + ", enqueueTime=" + filter.getEnqueueTime());
+ long start = System.currentTimeMillis();
+ EventHubClient eventHubClient = EventHubClient.create(connectionString, entityName);
+ if(filter.getOffset() != null) {
+ receiver = eventHubClient.getDefaultConsumerGroup().createReceiver(partitionId, filter.getOffset(), defaultCredits);
+ }
+ else if(filter.getEnqueueTime() != 0) {
+ receiver = eventHubClient.getDefaultConsumerGroup().createReceiver(partitionId, filter.getEnqueueTime(), defaultCredits);
+ }
+ else {
+ logger.error("Invalid IEventHubReceiverFilter, use default offset as filter");
+ receiver = eventHubClient.getDefaultConsumerGroup().createReceiver(partitionId, Constants.DefaultStartingOffset, defaultCredits);
+ }
+ long end = System.currentTimeMillis();
+ logger.info("created eventhub receiver, time taken(ms): " + (end-start));
+ }
+
+ @Override
+ public void close() {
+ if(receiver != null) {
+ receiver.close();
+ logger.info("closed eventhub receiver: partitionId=" + partitionId );
+ receiver = null;
+ }
+ }
+
+ @Override
+ public boolean isOpen() {
+ return (receiver != null);
+ }
+
+ @Override
+ public EventData receive(long timeoutInMilliseconds) {
+ long start = System.currentTimeMillis();
+ Message message = receiver.receive(timeoutInMilliseconds);
+ long end = System.currentTimeMillis();
+ long millis = (end - start);
+ receiveApiLatencyMean.update(millis);
+ receiveApiCallCount.incr();
+
+ if (message == null) {
+ return null;
+ }
+ receiveMessageCount.incr();
+
+ //logger.info(String.format("received a message. PartitionId: %s, Offset: %s", partitionId, this.lastOffset));
+ MessageId messageId = createMessageId(message);
+
+ return EventData.create(message, messageId);
+ }
+
+ private MessageId createMessageId(Message message) {
+ String offset = null;
+ long sequenceNumber = 0;
+
+ for (Section section : message.getPayload()) {
+ if (section instanceof MessageAnnotations) {
+ MessageAnnotations annotations = (MessageAnnotations) section;
+ HashMap annonationMap = (HashMap) annotations.getValue();
+
+ if (annonationMap.containsKey(OffsetKey)) {
+ offset = (String) annonationMap.get(OffsetKey);
+ }
+
+ if (annonationMap.containsKey(SequenceNumberKey)) {
+ sequenceNumber = (Long) annonationMap.get(SequenceNumberKey);
+ }
+ }
+ }
+
+ return MessageId.create(partitionId, offset, sequenceNumber);
+ }
+
+ @Override
+ public Map getMetricsData() {
+ Map ret = new HashMap();
+ ret.put(partitionId + "/receiveApiLatencyMean", receiveApiLatencyMean.getValueAndReset());
+ ret.put(partitionId + "/receiveApiCallCount", receiveApiCallCount.getValueAndReset());
+ ret.put(partitionId + "/receiveMessageCount", receiveMessageCount.getValueAndReset());
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java
new file mode 100755
index 0000000..9290e6e
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java
@@ -0,0 +1,258 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import backtype.storm.Config;
+import backtype.storm.metric.api.IMetric;
+import backtype.storm.spout.SpoutOutputCollector;
+import backtype.storm.task.TopologyContext;
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.topology.base.BaseRichSpout;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.qpid.amqp_1_0.client.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EventHubSpout extends BaseRichSpout {
+
+ private static final Logger logger = LoggerFactory.getLogger(EventHubSpout.class);
+
+ private final UUID instanceId;
+ private final EventHubSpoutConfig eventHubConfig;
+ private final IEventDataScheme scheme;
+ private final int checkpointIntervalInSeconds;
+
+ private IStateStore stateStore;
+ private IPartitionCoordinator partitionCoordinator;
+ private IPartitionManagerFactory pmFactory;
+ private IEventHubReceiverFactory recvFactory;
+ private SpoutOutputCollector collector;
+ private long lastCheckpointTime;
+ private int currentPartitionIndex = -1;
+
+ public EventHubSpout(EventHubSpoutConfig spoutConfig) {
+ this(spoutConfig, null, null, null);
+ }
+
+ public EventHubSpout(EventHubSpoutConfig spoutConfig,
+ IStateStore store,
+ IPartitionManagerFactory pmFactory,
+ IEventHubReceiverFactory recvFactory) {
+ this.eventHubConfig = spoutConfig;
+ this.scheme = spoutConfig.getEventDataScheme();
+ this.instanceId = UUID.randomUUID();
+ this.checkpointIntervalInSeconds = spoutConfig.getCheckpointIntervalInSeconds();
+ this.lastCheckpointTime = System.currentTimeMillis();
+ stateStore = store;
+ this.pmFactory = pmFactory;
+ if(this.pmFactory == null) {
+ this.pmFactory = new IPartitionManagerFactory() {
+ @Override
+ public IPartitionManager create(EventHubSpoutConfig spoutConfig,
+ String partitionId, IStateStore stateStore,
+ IEventHubReceiver receiver) {
+ return new PartitionManager(spoutConfig, partitionId,
+ stateStore, receiver);
+ }
+ };
+ }
+ this.recvFactory = recvFactory;
+ if(this.recvFactory == null) {
+ this.recvFactory = new IEventHubReceiverFactory() {
+ @Override
+ public IEventHubReceiver create(EventHubSpoutConfig spoutConfig,
+ String partitionId) {
+ return new EventHubReceiverImpl(spoutConfig, partitionId);
+ }
+ };
+ }
+
+ }
+
+ /**
+ * This is a extracted method that is easy to test
+ * @param config
+ * @param totalTasks
+ * @param taskIndex
+ * @param collector
+ * @throws Exception
+ */
+ public void preparePartitions(Map config, int totalTasks, int taskIndex, SpoutOutputCollector collector) throws Exception {
+ this.collector = collector;
+ if(stateStore == null) {
+ String zkEndpointAddress = eventHubConfig.getZkConnectionString();
+ if (zkEndpointAddress == null || zkEndpointAddress.length() == 0) {
+ //use storm's zookeeper servers if not specified.
+ @SuppressWarnings("unchecked")
+ List<String> zkServers = (List<String>) config.get(Config.STORM_ZOOKEEPER_SERVERS);
+ Integer zkPort = ((Number) config.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
+ StringBuilder sb = new StringBuilder();
+ for (String zk : zkServers) {
+ if (sb.length() > 0) {
+ sb.append(',');
+ }
+ sb.append(zk+":"+zkPort);
+ }
+ zkEndpointAddress = sb.toString();
+ }
+ stateStore = new ZookeeperStateStore(zkEndpointAddress,
+ (Integer)config.get(Config.STORM_ZOOKEEPER_RETRY_TIMES),
+ (Integer)config.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL));
+ }
+ stateStore.open();
+
+ partitionCoordinator = new StaticPartitionCoordinator(
+ eventHubConfig, taskIndex, totalTasks, stateStore, pmFactory, recvFactory);
+
+ for (IPartitionManager partitionManager :
+ partitionCoordinator.getMyPartitionManagers()) {
+ partitionManager.open();
+ }
+ }
+
+ @Override
+ public void open(Map config, TopologyContext context, SpoutOutputCollector collector) {
+ logger.info("begin: open()");
+ String topologyName = (String) config.get(Config.TOPOLOGY_NAME);
+ eventHubConfig.setTopologyName(topologyName);
+
+ int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();
+ int taskIndex = context.getThisTaskIndex();
+ if (totalTasks > eventHubConfig.getPartitionCount()) {
+ throw new RuntimeException("total tasks of EventHubSpout is greater than partition count.");
+ }
+
+ logger.info(String.format("topologyName: %s, totalTasks: %d, taskIndex: %d", topologyName, totalTasks, taskIndex));
+
+ try {
+ preparePartitions(config, totalTasks, taskIndex, collector);
+ } catch (Exception e) {
+ logger.error(e.getMessage());
+ throw new RuntimeException(e);
+ }
+
+ //register metrics
+ context.registerMetric("EventHubReceiver", new IMetric() {
+ @Override
+ public Object getValueAndReset() {
+ Map concatMetricsDataMaps = new HashMap();
+ for (IPartitionManager partitionManager :
+ partitionCoordinator.getMyPartitionManagers()) {
+ concatMetricsDataMaps.putAll(partitionManager.getMetricsData());
+ }
+ return concatMetricsDataMaps;
+ }
+ }, (Integer)config.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS));
+ logger.info("end open()");
+ }
+
+ @Override
+ public void nextTuple() {
+ EventData eventData = null;
+
+ List<IPartitionManager> partitionManagers = partitionCoordinator.getMyPartitionManagers();
+ for (int i = 0; i < partitionManagers.size(); i++) {
+ currentPartitionIndex = (currentPartitionIndex + 1) % partitionManagers.size();
+ IPartitionManager partitionManager = partitionManagers.get(currentPartitionIndex);
+
+ if (partitionManager == null) {
+ throw new RuntimeException("partitionManager doesn't exist.");
+ }
+
+ eventData = partitionManager.receive();
+
+ if (eventData != null) {
+ break;
+ }
+ }
+
+
+ if (eventData != null) {
+ MessageId messageId = eventData.getMessageId();
+ Message message = eventData.getMessage();
+
+ List<Object> tuples = scheme.deserialize(message);
+
+ if (tuples != null) {
+ collector.emit(tuples, messageId);
+ }
+ }
+
+ checkpointIfNeeded();
+
+ // We don't need to sleep here because the IPartitionManager.receive() is
+ // a blocked call so it's fine to call this function in a tight loop.
+ }
+
+ @Override
+ public void ack(Object msgId) {
+ MessageId messageId = (MessageId) msgId;
+ IPartitionManager partitionManager = partitionCoordinator.getPartitionManager(messageId.getPartitionId());
+ String offset = messageId.getOffset();
+ partitionManager.ack(offset);
+ }
+
+ @Override
+ public void fail(Object msgId) {
+ MessageId messageId = (MessageId) msgId;
+ IPartitionManager partitionManager = partitionCoordinator.getPartitionManager(messageId.getPartitionId());
+ String offset = messageId.getOffset();
+ partitionManager.fail(offset);
+ }
+
+ @Override
+ public void deactivate() {
+ // let's checkpoint so that we can get the last checkpoint when restarting.
+ checkpoint();
+ }
+
+ @Override
+ public void close() {
+ for (IPartitionManager partitionManager :
+ partitionCoordinator.getMyPartitionManagers()) {
+ partitionManager.close();
+ }
+ stateStore.close();
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(scheme.getOutputFields());
+ }
+
+ private void checkpointIfNeeded() {
+ long nextCheckpointTime = lastCheckpointTime + checkpointIntervalInSeconds * 1000;
+ if (nextCheckpointTime < System.currentTimeMillis()) {
+
+ checkpoint();
+ lastCheckpointTime = System.currentTimeMillis();
+ }
+ }
+
+ private void checkpoint() {
+ for (IPartitionManager partitionManager :
+ partitionCoordinator.getMyPartitionManagers()) {
+ partitionManager.checkpoint();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java
new file mode 100755
index 0000000..ae11680
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java
@@ -0,0 +1,165 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.util.ArrayList;
+import java.util.List;
+
+public class EventHubSpoutConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private final String userName;
+ private final String password;
+ private final String namespace;
+ private final String entityPath;
+ private final String zkConnectionString;
+ private final int partitionCount;
+ private final int checkpointIntervalInSeconds;
+ private final int receiverCredits;
+ private final int maxPendingMsgsPerPartition;
+ private final long enqueueTimeFilter; //timestamp in millisecond
+
+ private String connectionString;
+ private String targetFqnAddress;
+ private String topologyName;
+ private IEventDataScheme scheme;
+
+ public EventHubSpoutConfig(String username, String password, String namespace,
+ String entityPath, int partitionCount, String zkConnectionString) {
+ this(username, password, namespace, entityPath, partitionCount,
+ zkConnectionString, 10, 1024, 1024, 0);
+ }
+
+ //Keep this constructor for backward compatibility
+ public EventHubSpoutConfig(String username, String password, String namespace,
+ String entityPath, int partitionCount, String zkConnectionString,
+ int checkpointIntervalInSeconds, int receiverCredits) {
+ this(username, password, namespace, entityPath, partitionCount,
+ zkConnectionString, checkpointIntervalInSeconds, receiverCredits, 1024, 0);
+ }
+
+ public EventHubSpoutConfig(String username, String password, String namespace,
+ String entityPath, int partitionCount, String zkConnectionString,
+ int checkpointIntervalInSeconds, int receiverCredits, int maxPendingMsgsPerPartition, long enqueueTimeFilter) {
+ this.userName = username;
+ this.password = password;
+ this.connectionString = buildConnectionString(username, password, namespace);
+ this.namespace = namespace;
+ this.entityPath = entityPath;
+ this.partitionCount = partitionCount;
+ this.zkConnectionString = zkConnectionString;
+ this.checkpointIntervalInSeconds = checkpointIntervalInSeconds;
+ this.receiverCredits = receiverCredits;
+ this.maxPendingMsgsPerPartition = maxPendingMsgsPerPartition;
+ this.enqueueTimeFilter = enqueueTimeFilter;
+ this.scheme = new EventDataScheme();
+ }
+
+ public String getConnectionString() {
+ return connectionString;
+ }
+
+ public String getNamespace() {
+ return namespace;
+ }
+
+ public String getEntityPath() {
+ return entityPath;
+ }
+
+ public String getZkConnectionString() {
+ return zkConnectionString;
+ }
+
+ public int getCheckpointIntervalInSeconds() {
+ return checkpointIntervalInSeconds;
+ }
+
+ public int getPartitionCount() {
+ return partitionCount;
+ }
+
+ public int getReceiverCredits() {
+ return receiverCredits;
+ }
+
+ public int getMaxPendingMsgsPerPartition() {
+ return maxPendingMsgsPerPartition;
+ }
+
+ public long getEnqueueTimeFilter() {
+ return enqueueTimeFilter;
+ }
+
+ public String getTopologyName() {
+ return topologyName;
+ }
+
+ public void setTopologyName(String value) {
+ topologyName = value;
+ }
+
+ public IEventDataScheme getEventDataScheme() {
+ return scheme;
+ }
+
+ public void setEventDataScheme(IEventDataScheme scheme) {
+ this.scheme = scheme;
+ }
+
+ public List<String> getPartitionList() {
+ List<String> partitionList = new ArrayList<String>();
+
+ for (int i = 0; i < this.partitionCount; i++) {
+ partitionList.add(Integer.toString(i));
+ }
+
+ return partitionList;
+ }
+
+ public void setTargetAddress(String targetFqnAddress) {
+ this.targetFqnAddress = targetFqnAddress;
+ this.connectionString = buildConnectionString(
+ this.userName, this.password, this.namespace, this.targetFqnAddress);
+ }
+
+ public static String buildConnectionString(String username, String password, String namespace) {
+ String targetFqnAddress = "servicebus.windows.net";
+ return buildConnectionString(username, password, namespace, targetFqnAddress);
+ }
+
+ public static String buildConnectionString(String username, String password,
+ String namespace, String targetFqnAddress) {
+ return "amqps://" + username + ":" + encodeString(password)
+ + "@" + namespace + "." + targetFqnAddress;
+ }
+
+ private static String encodeString(String input) {
+ try {
+ return URLEncoder.encode(input, "UTF-8");
+ } catch (UnsupportedEncodingException e) {
+ //We don't need to throw this exception because the exception won't
+ //happen because of user input. Our unit tests will catch this error.
+ return "";
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutException.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutException.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutException.java
new file mode 100755
index 0000000..0fd6ac4
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutException.java
@@ -0,0 +1,37 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+public class EventHubSpoutException extends Exception {
+
+ public EventHubSpoutException() {
+ super();
+ }
+
+ public EventHubSpoutException(String message) {
+ super(message);
+ }
+
+ public EventHubSpoutException(Throwable cause) {
+ super(cause);
+ }
+
+ public EventHubSpoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java
new file mode 100755
index 0000000..bd655d6
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java
@@ -0,0 +1,25 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+public class FieldConstants {
+
+ public static final String PartitionKey = "partitionKey";
+ public static final String Offset = "offset";
+ public static final String Message = "message";
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java
new file mode 100755
index 0000000..c96767d
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java
@@ -0,0 +1,30 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import backtype.storm.tuple.Fields;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.qpid.amqp_1_0.client.Message;
+
+public interface IEventDataScheme extends Serializable {
+
+ List<Object> deserialize(Message message);
+
+ Fields getOutputFields();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java
new file mode 100755
index 0000000..45e9e57
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java
@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.util.Map;
+
+import org.apache.storm.eventhubs.client.EventHubException;
+
+public interface IEventHubReceiver {
+
+ void open(IEventHubReceiverFilter filter) throws EventHubException;
+
+ void close();
+
+ boolean isOpen();
+
+ EventData receive(long timeoutInMilliseconds);
+
+ Map getMetricsData();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java
new file mode 100755
index 0000000..fbebdc8
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java
@@ -0,0 +1,30 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+/**
+ *
+ */
+package org.apache.storm.eventhubs.spout;
+
+import java.io.Serializable;
+
+/**
+ * An abstract factory to generate EventHubReceiver
+ */
+public interface IEventHubReceiverFactory extends Serializable {
+ IEventHubReceiver create(EventHubSpoutConfig config, String partitionId);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFilter.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFilter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFilter.java
new file mode 100755
index 0000000..e5b93cf
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFilter.java
@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+/**
+ * The filter to create an EventHubs receiver
+ */
+public interface IEventHubReceiverFilter {
+ /**
+ * Get offset to filter events based on offset
+ * @return null if offset not set
+ */
+ String getOffset();
+
+ /**
+ * Get timestamp to filter events based on enqueue time.
+ * @return 0 if enqueue time is not set
+ */
+ long getEnqueueTime();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java
new file mode 100755
index 0000000..a460681
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java
@@ -0,0 +1,27 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.util.List;
+
+public interface IPartitionCoordinator {
+
+ List<IPartitionManager> getMyPartitionManagers();
+
+ IPartitionManager getPartitionManager(String partitionId);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java
new file mode 100755
index 0000000..ac986d3
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java
@@ -0,0 +1,37 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.util.Map;
+
+public interface IPartitionManager {
+
+ void open() throws Exception;
+
+ void close();
+
+ EventData receive();
+
+ void checkpoint();
+
+ void ack(String offset);
+
+ void fail(String offset);
+
+ Map getMetricsData();
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java
new file mode 100755
index 0000000..dc136eb
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java
@@ -0,0 +1,33 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+/**
+ *
+ */
+package org.apache.storm.eventhubs.spout;
+
+import java.io.Serializable;
+
+/**
+ * An interface of factory method to create IPartitionManager
+ */
+public interface IPartitionManagerFactory extends Serializable {
+ IPartitionManager create(EventHubSpoutConfig spoutConfig,
+ String partitionId,
+ IStateStore stateStore,
+ IEventHubReceiver receiver);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java
new file mode 100755
index 0000000..f0ee2be
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java
@@ -0,0 +1,31 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.io.Serializable;
+
+public interface IStateStore extends Serializable {
+
+ public void open();
+
+ public void close();
+
+ public void saveData(String path, String data);
+
+ public String readData(String path);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java
new file mode 100755
index 0000000..2247f1f
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java
@@ -0,0 +1,56 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+public class MessageId {
+
+ private final String partitionId;
+ private final String offset;
+ private final long sequenceNumber;
+
+ public MessageId(
+ String partitionId,
+ String offset,
+ long sequenceNumber) {
+ this.partitionId = partitionId;
+ this.offset = offset;
+ this.sequenceNumber = sequenceNumber;
+ }
+
+ public static MessageId create(String partitionId, String offset, long sequenceNumber) {
+ return new MessageId(partitionId, offset, sequenceNumber);
+ }
+
+ public String getPartitionId() {
+ return this.partitionId;
+ }
+
+ public String getOffset() {
+ return this.offset;
+ }
+
+ public Long getSequenceNumber() {
+ return this.sequenceNumber;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("PartitionId: %s, Offset: %s, SequenceNumber: %s",
+ this.partitionId, this.offset, this.sequenceNumber);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java
new file mode 100755
index 0000000..1054742
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java
@@ -0,0 +1,101 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.TreeSet;
+
+public class PartitionManager extends SimplePartitionManager {
+ private static final Logger logger = LoggerFactory.getLogger(PartitionManager.class);
+ private final int ehReceiveTimeoutMs = 5000;
+
+ //all sent events are stored in pending
+ private final Map<String, EventData> pending;
+ //all failed events are put in toResend, which is sorted by event's offset
+ private final TreeSet<EventData> toResend;
+
+ public PartitionManager(
+ EventHubSpoutConfig spoutConfig,
+ String partitionId,
+ IStateStore stateStore,
+ IEventHubReceiver receiver) {
+
+ super(spoutConfig, partitionId, stateStore, receiver);
+
+ this.pending = new LinkedHashMap<String, EventData>();
+ this.toResend = new TreeSet<EventData>();
+ }
+
+ @Override
+ public EventData receive() {
+ if(pending.size() >= config.getMaxPendingMsgsPerPartition()) {
+ return null;
+ }
+
+ EventData eventData;
+ if (toResend.isEmpty()) {
+ eventData = receiver.receive(ehReceiveTimeoutMs);
+ } else {
+ eventData = toResend.pollFirst();
+ }
+
+ if (eventData != null) {
+ lastOffset = eventData.getMessageId().getOffset();
+ pending.put(lastOffset, eventData);
+ }
+
+ return eventData;
+ }
+
+ @Override
+ public void ack(String offset) {
+ pending.remove(offset);
+ }
+
+ @Override
+ public void fail(String offset) {
+ logger.warn("fail on " + offset);
+ EventData eventData = pending.remove(offset);
+ toResend.add(eventData);
+ }
+
+ @Override
+ protected String getCompletedOffset() {
+ String offset = null;
+
+ if(pending.size() > 0) {
+ //find the smallest offset in pending list
+ offset = pending.keySet().iterator().next();
+ }
+ if(toResend.size() > 0) {
+ //find the smallest offset in toResend list
+ String offset2 = toResend.first().getMessageId().getOffset();
+ if(offset == null || offset2.compareTo(offset) < 0) {
+ offset = offset2;
+ }
+ }
+ if(offset == null) {
+ offset = lastOffset;
+ }
+ return offset;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java
new file mode 100755
index 0000000..bcbcbac
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java
@@ -0,0 +1,136 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.storm.eventhubs.client.Constants;
+
+/**
+ * A simple partition manager that does not re-send failed messages
+ */
+public class SimplePartitionManager implements IPartitionManager {
+ private static final Logger logger = LoggerFactory.getLogger(SimplePartitionManager.class);
+ private static final String statePathPrefix = "/eventhubspout";
+
+ protected final IEventHubReceiver receiver;
+ protected String lastOffset = "-1";
+ protected String committedOffset = "-1";
+
+ protected final EventHubSpoutConfig config;
+ private final String partitionId;
+ private final IStateStore stateStore;
+ private final String statePath;
+
+ public SimplePartitionManager(
+ EventHubSpoutConfig spoutConfig,
+ String partitionId,
+ IStateStore stateStore,
+ IEventHubReceiver receiver) {
+ this.receiver = receiver;
+ this.config = spoutConfig;
+ this.partitionId = partitionId;
+ this.statePath = this.getPartitionStatePath();
+ this.stateStore = stateStore;
+ }
+
+ @Override
+ public void open() throws Exception {
+
+ //read from state store, if not found, use startingOffset
+ String offset = stateStore.readData(statePath);
+ logger.info("read offset from state store: " + offset);
+ if(offset == null) {
+ offset = Constants.DefaultStartingOffset;
+ }
+
+ EventHubReceiverFilter filter = new EventHubReceiverFilter();
+ if (offset.equals(Constants.DefaultStartingOffset)
+ && config.getEnqueueTimeFilter() != 0) {
+ filter.setEnqueueTime(config.getEnqueueTimeFilter());
+ }
+ else {
+ filter.setOffset(offset);
+ }
+
+ receiver.open(filter);
+ }
+
+ @Override
+ public void close() {
+ this.receiver.close();
+ this.checkpoint();
+ }
+
+ @Override
+ public void checkpoint() {
+ String completedOffset = getCompletedOffset();
+ if(!committedOffset.equals(completedOffset)) {
+ logger.info("saving state " + completedOffset);
+ stateStore.saveData(statePath, completedOffset);
+ committedOffset = completedOffset;
+ }
+ }
+
+ protected String getCompletedOffset() {
+ return lastOffset;
+ }
+
+ @Override
+ public EventData receive() {
+ EventData eventData = receiver.receive(5000);
+ if (eventData != null) {
+ lastOffset = eventData.getMessageId().getOffset();
+ }
+ return eventData;
+ }
+
+ @Override
+ public void ack(String offset) {
+ //do nothing
+ }
+
+ @Override
+ public void fail(String offset) {
+ logger.warn("fail on " + offset);
+ //do nothing
+ }
+
+ private String getPartitionStatePath() {
+
+ // Partition state path =
+ // "/{prefix}/{topologyName}/{namespace}/{entityPath}/partitions/{partitionId}/state";
+ String namespace = config.getNamespace();
+ String entityPath = config.getEntityPath();
+ String topologyName = config.getTopologyName();
+
+ String partitionStatePath = statePathPrefix + "/" + topologyName + "/" + namespace + "/" + entityPath + "/partitions/" + this.partitionId;
+
+ logger.info("partition state path: " + partitionStatePath);
+
+ return partitionStatePath;
+ }
+
+ @Override
+ public Map getMetricsData() {
+ return receiver.getMetricsData();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java
new file mode 100755
index 0000000..3f5f156
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java
@@ -0,0 +1,85 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.storm.eventhubs.client.Constants;
+
+public class StaticPartitionCoordinator implements IPartitionCoordinator {
+
+ private static final Logger logger = LoggerFactory.getLogger(StaticPartitionCoordinator.class);
+
+ protected final EventHubSpoutConfig config;
+ protected final int taskIndex;
+ protected final int totalTasks;
+ protected final List<IPartitionManager> partitionManagers;
+ protected final Map<String, IPartitionManager> partitionManagerMap;
+ protected final IStateStore stateStore;
+
+ public StaticPartitionCoordinator(
+ EventHubSpoutConfig spoutConfig,
+ int taskIndex,
+ int totalTasks,
+ IStateStore stateStore,
+ IPartitionManagerFactory pmFactory,
+ IEventHubReceiverFactory recvFactory) {
+
+ this.config = spoutConfig;
+ this.taskIndex = taskIndex;
+ this.totalTasks = totalTasks;
+ this.stateStore = stateStore;
+ List<String> partitionIds = calculateParititionIdsToOwn();
+ partitionManagerMap = new HashMap<String, IPartitionManager>();
+ partitionManagers = new ArrayList<IPartitionManager>();
+
+ for (String partitionId : partitionIds) {
+ IEventHubReceiver receiver = recvFactory.create(config, partitionId);
+ IPartitionManager partitionManager = pmFactory.create(
+ config, partitionId, stateStore, receiver);
+ partitionManagerMap.put(partitionId, partitionManager);
+ partitionManagers.add(partitionManager);
+ }
+ }
+
+ @Override
+ public List<IPartitionManager> getMyPartitionManagers() {
+ return partitionManagers;
+ }
+
+ @Override
+ public IPartitionManager getPartitionManager(String partitionId) {
+ return partitionManagerMap.get(partitionId);
+ }
+
+ protected List<String> calculateParititionIdsToOwn() {
+ List<String> taskPartitions = new ArrayList<String>();
+ for (int i = this.taskIndex; i < config.getPartitionCount(); i += this.totalTasks) {
+ taskPartitions.add(Integer.toString(i));
+ logger.info(String.format("taskIndex %d owns partitionId %d.", this.taskIndex, i));
+ }
+
+ return taskPartitions;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java
new file mode 100755
index 0000000..6af9df6
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java
@@ -0,0 +1,95 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.spout;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ZookeeperStateStore implements IStateStore {
+ private static final long serialVersionUID = 1L;
+ private static final Logger logger = LoggerFactory.getLogger(ZookeeperStateStore.class);
+
+ private final String zookeeperConnectionString;
+ private final CuratorFramework curatorFramework;
+
+ public ZookeeperStateStore(String zookeeperConnectionString) {
+ this(zookeeperConnectionString, 3, 100);
+ }
+
+ public ZookeeperStateStore(String connectionString, int retries, int retryInterval) {
+ if (connectionString == null) {
+ zookeeperConnectionString = "localhost:2181";
+ } else {
+ zookeeperConnectionString = connectionString;
+ }
+
+ RetryPolicy retryPolicy = new RetryNTimes(retries, retryInterval);
+ curatorFramework = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
+ }
+
+ @Override
+ public void open() {
+ curatorFramework.start();
+ }
+
+ @Override
+ public void close() {
+ curatorFramework.close();
+ }
+
+ @Override
+ public void saveData(String statePath, String data) {
+ data = data == null ? "" : data;
+ byte[] bytes = data.getBytes();
+
+ try {
+ if (curatorFramework.checkExists().forPath(statePath) == null) {
+ curatorFramework.create().creatingParentsIfNeeded().forPath(statePath, bytes);
+ } else {
+ curatorFramework.setData().forPath(statePath, bytes);
+ }
+
+ logger.info(String.format("data was saved. path: %s, data: %s.", statePath, data));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String readData(String statePath) {
+ try {
+ if (curatorFramework.checkExists().forPath(statePath) == null) {
+ // do we want to throw an exception if path doesn't exist??
+ return null;
+ } else {
+ byte[] bytes = curatorFramework.getData().forPath(statePath);
+ String data = new String(bytes);
+
+ logger.info(String.format("data was retrieved. path: %s, data: %s.", statePath, data));
+
+ return data;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java
new file mode 100755
index 0000000..a43193d
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java
@@ -0,0 +1,60 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import storm.trident.spout.IOpaquePartitionedTridentSpout;
+import storm.trident.spout.IPartitionedTridentSpout;
+
+import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
+
+public class Coordinator implements IPartitionedTridentSpout.Coordinator<Partitions>,
+ IOpaquePartitionedTridentSpout.Coordinator<Partitions> {
+ private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);
+ private final EventHubSpoutConfig spoutConfig;
+ Partitions partitions;
+
+ public Coordinator(EventHubSpoutConfig spoutConfig) {
+ this.spoutConfig = spoutConfig;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public Partitions getPartitionsForBatch() {
+ if(partitions != null) {
+ return partitions;
+ }
+
+ partitions = new Partitions();
+ for(int i=0; i<spoutConfig.getPartitionCount(); ++i) {
+ partitions.addPartition(new Partition(spoutConfig, Integer.toString(i)));
+ }
+ logger.info("created partitions, size=" + spoutConfig.getPartitionCount());
+ return partitions;
+ }
+
+ @Override
+ public boolean isReady(long txid) {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java
new file mode 100755
index 0000000..fbe779d
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java
@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.util.List;
+
+import org.apache.storm.eventhubs.spout.EventData;
+
+public interface ITridentPartitionManager {
+ boolean open(String offset);
+ void close();
+
+ /**
+ * receive a batch of messages from EvenHub up to "count" messages
+ * @param offset the starting offset
+ * @param count max number of messages in this batch
+ * @return list of EventData, if failed to receive, return empty list
+ */
+ public List<EventData> receiveBatch(String offset, int count);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java
new file mode 100755
index 0000000..5804e28
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java
@@ -0,0 +1,26 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.io.Serializable;
+
+import org.apache.storm.eventhubs.spout.IEventHubReceiver;
+
+public interface ITridentPartitionManagerFactory extends Serializable {
+ ITridentPartitionManager create(IEventHubReceiver receiver);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
new file mode 100755
index 0000000..c7bd8c3
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
@@ -0,0 +1,69 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
+import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
+
+import storm.trident.operation.TridentCollector;
+import storm.trident.spout.IOpaquePartitionedTridentSpout;
+import storm.trident.topology.TransactionAttempt;
+
+/**
+ * A thin wrapper of TransactionalTridentEventHubEmitter for OpaqueTridentEventHubSpout
+ */
+public class OpaqueTridentEventHubEmitter implements IOpaquePartitionedTridentSpout.Emitter<Partitions, Partition, Map> {
+ private final TransactionalTridentEventHubEmitter transactionalEmitter;
+ public OpaqueTridentEventHubEmitter(EventHubSpoutConfig spoutConfig) {
+ transactionalEmitter = new TransactionalTridentEventHubEmitter(spoutConfig);
+ }
+
+ public OpaqueTridentEventHubEmitter(EventHubSpoutConfig spoutConfig,
+ int batchSize,
+ ITridentPartitionManagerFactory pmFactory,
+ IEventHubReceiverFactory recvFactory) {
+ transactionalEmitter = new TransactionalTridentEventHubEmitter(spoutConfig,
+ batchSize,
+ pmFactory,
+ recvFactory);
+ }
+
+ @Override
+ public void close() {
+ transactionalEmitter.close();
+ }
+
+ @Override
+ public Map emitPartitionBatch(TransactionAttempt attempt, TridentCollector collector,
+ Partition partition, Map meta) {
+ return transactionalEmitter.emitPartitionBatchNew(attempt, collector, partition, meta);
+ }
+
+ @Override
+ public List<Partition> getOrderedPartitions(Partitions partitions) {
+ return transactionalEmitter.getOrderedPartitions(partitions);
+ }
+
+ @Override
+ public void refreshPartitions(List<Partition> partitionList) {
+ transactionalEmitter.refreshPartitions(partitionList);
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java
new file mode 100755
index 0000000..46084ef
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java
@@ -0,0 +1,64 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.util.Map;
+
+import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
+import org.apache.storm.eventhubs.spout.IEventDataScheme;
+
+import backtype.storm.task.TopologyContext;
+import backtype.storm.tuple.Fields;
+import storm.trident.spout.IOpaquePartitionedTridentSpout;
+
+/**
+ * Opaque Trident EventHubs Spout
+ */
+public class OpaqueTridentEventHubSpout implements IOpaquePartitionedTridentSpout<Partitions, Partition, Map> {
+ private static final long serialVersionUID = 1L;
+ private final IEventDataScheme scheme;
+ private final EventHubSpoutConfig spoutConfig;
+
+ public OpaqueTridentEventHubSpout(EventHubSpoutConfig config) {
+ spoutConfig = config;
+ scheme = spoutConfig.getEventDataScheme();
+ }
+
+ @Override
+ public Map getComponentConfiguration() {
+ return null;
+ }
+
+ @Override
+ public IOpaquePartitionedTridentSpout.Coordinator<Partitions> getCoordinator(
+ Map conf, TopologyContext context) {
+ return new org.apache.storm.eventhubs.trident.Coordinator(spoutConfig);
+ }
+
+ @Override
+ public IOpaquePartitionedTridentSpout.Emitter<Partitions, Partition, Map> getEmitter(
+ Map conf, TopologyContext context) {
+ return new OpaqueTridentEventHubEmitter(spoutConfig);
+ }
+
+ @Override
+ public Fields getOutputFields() {
+ return scheme.getOutputFields();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java
new file mode 100755
index 0000000..8b166cd
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java
@@ -0,0 +1,39 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.io.Serializable;
+import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
+import storm.trident.spout.ISpoutPartition;
+
+/**
+ * Represents an EventHub partition
+ */
+public class Partition implements ISpoutPartition, Serializable {
+ private static final long serialVersionUID = 1L;
+ String partitionId;
+
+ public Partition(EventHubSpoutConfig config, String partitionId) {
+ this.partitionId = partitionId;
+ }
+
+ @Override
+ public String getId() {
+ return partitionId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java
new file mode 100755
index 0000000..235f5b6
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java
@@ -0,0 +1,41 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Represents all EventHub partitions a spout is receiving messages from.
+ */
+public class Partitions implements Serializable {
+ private static final long serialVersionUID = 1L;
+ private List<Partition> partitionList;
+ public Partitions() {
+ partitionList = new ArrayList<Partition>();
+ }
+
+ public void addPartition(Partition partition) {
+ partitionList.add(partition);
+ }
+
+ public List<Partition> getPartitions() {
+ return partitionList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/eff6cfa4/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java
new file mode 100755
index 0000000..2b92c3c
--- /dev/null
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java
@@ -0,0 +1,167 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *******************************************************************************/
+package org.apache.storm.eventhubs.trident;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.storm.eventhubs.spout.EventData;
+import org.apache.storm.eventhubs.spout.EventHubReceiverImpl;
+import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
+import org.apache.storm.eventhubs.spout.IEventHubReceiver;
+import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
+import org.apache.storm.eventhubs.client.Constants;
+
+import storm.trident.operation.TridentCollector;
+import storm.trident.spout.IOpaquePartitionedTridentSpout;
+import storm.trident.spout.IPartitionedTridentSpout;
+import storm.trident.topology.TransactionAttempt;
+
+
+public class TransactionalTridentEventHubEmitter
+ implements IPartitionedTridentSpout.Emitter<Partitions, Partition, Map> {
+ private static final Logger logger = LoggerFactory.getLogger(TransactionalTridentEventHubEmitter.class);
+ private final int batchSize;
+ private final EventHubSpoutConfig spoutConfig;
+ private Map<String, ITridentPartitionManager> pmMap;
+ private ITridentPartitionManagerFactory pmFactory;
+ private IEventHubReceiverFactory recvFactory;
+
+ public TransactionalTridentEventHubEmitter(EventHubSpoutConfig spoutConfig) {
+ //use batch size that matches the default credit size
+ this(spoutConfig, spoutConfig.getReceiverCredits(), null, null);
+ }
+
+ public TransactionalTridentEventHubEmitter(final EventHubSpoutConfig spoutConfig,
+ int batchSize,
+ ITridentPartitionManagerFactory pmFactory,
+ IEventHubReceiverFactory recvFactory) {
+ this.spoutConfig = spoutConfig;
+ this.batchSize = batchSize;
+ this.pmFactory = pmFactory;
+ this.recvFactory = recvFactory;
+ pmMap = new HashMap<String, ITridentPartitionManager>();
+ if(this.pmFactory == null) {
+ this.pmFactory = new ITridentPartitionManagerFactory() {
+ @Override
+ public ITridentPartitionManager create(IEventHubReceiver receiver) {
+ return new TridentPartitionManager(spoutConfig, receiver);
+ }
+ };
+ }
+ if(this.recvFactory == null) {
+ this.recvFactory = new IEventHubReceiverFactory() {
+ @Override
+ public IEventHubReceiver create(EventHubSpoutConfig config,
+ String partitionId) {
+ return new EventHubReceiverImpl(config, partitionId);
+ }
+ };
+ }
+ }
+
+ @Override
+ public void close() {
+ for(ITridentPartitionManager pm: pmMap.values()) {
+ pm.close();
+ }
+ }
+
+ /**
+ * Check if partition manager for a given partiton is created
+ * if not, create it.
+ * @param partition
+ */
+ private ITridentPartitionManager getOrCreatePartitionManager(Partition partition) {
+ ITridentPartitionManager pm;
+ if(!pmMap.containsKey(partition.getId())) {
+ IEventHubReceiver receiver = recvFactory.create(spoutConfig, partition.getId());
+ pm = pmFactory.create(receiver);
+ pmMap.put(partition.getId(), pm);
+ }
+ else {
+ pm = pmMap.get(partition.getId());
+ }
+ return pm;
+ }
+
+ @Override
+ public void emitPartitionBatch(TransactionAttempt attempt,
+ TridentCollector collector, Partition partition, Map meta) {
+ String offset = (String)meta.get("offset");
+ int count = Integer.parseInt((String)meta.get("count"));
+ logger.info("re-emit for partition " + partition.getId() + ", offset=" + offset + ", count=" + count);
+ ITridentPartitionManager pm = getOrCreatePartitionManager(partition);
+ List<EventData> listEvents = pm.receiveBatch(offset, count);
+ if(listEvents.size() != count) {
+ logger.error("failed to refetch eventhub messages, new count=" + listEvents.size());
+ return;
+ }
+
+ for(EventData ed: listEvents) {
+ List<Object> tuples =
+ spoutConfig.getEventDataScheme().deserialize(ed.getMessage());
+ collector.emit(tuples);
+ }
+ }
+
+ @Override
+ public Map emitPartitionBatchNew(TransactionAttempt attempt,
+ TridentCollector collector, Partition partition, Map meta) {
+ ITridentPartitionManager pm = getOrCreatePartitionManager(partition);
+ String offset = Constants.DefaultStartingOffset;
+ if(meta != null && meta.containsKey("nextOffset")) {
+ offset = (String)meta.get("nextOffset");
+ }
+ //logger.info("emit for partition " + partition.getId() + ", offset=" + offset);
+ String nextOffset = offset;
+
+ List<EventData> listEvents = pm.receiveBatch(offset, batchSize);
+
+ for(EventData ed: listEvents) {
+ //update nextOffset;
+ nextOffset = ed.getMessageId().getOffset();
+ List<Object> tuples =
+ spoutConfig.getEventDataScheme().deserialize(ed.getMessage());
+ collector.emit(tuples);
+ }
+ //logger.info("emitted new batches: " + listEvents.size());
+
+ Map newMeta = new HashMap();
+ newMeta.put("offset", offset);
+ newMeta.put("nextOffset", nextOffset);
+ newMeta.put("count", ""+listEvents.size());
+ return newMeta;
+ }
+
+ @Override
+ public List<Partition> getOrderedPartitions(Partitions partitions) {
+ return partitions.getPartitions();
+ }
+
+ @Override
+ public void refreshPartitions(List<Partition> partitionList) {
+ //partition info does not change in EventHub
+ return;
+ }
+
+}