You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by priyank5485 <gi...@git.apache.org> on 2016/07/23 07:10:37 UTC
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
GitHub user priyank5485 opened a pull request:
https://github.com/apache/storm/pull/1586
STORM-1839: Storm spout implementation for Amazon Kinesis Streams.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/priyank5485/storm STORM-1839
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1586.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1586
----
commit de68c267fcb7555c7729c9377d3f6d1e504ec25e
Author: Priyank <ps...@hortonworks.com>
Date: 2016-07-12T19:17:54Z
STORM-1839: Storm spout implementation for Amazon Kinesis Streams.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72298863
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ // add all fetched records to the set of failed records if they are present in failed set
+ for (Record record: records) {
+ KinesisMessageId current = new KinesisMessageId(kinesisMessageId.getStreamName(), kinesisMessageId.getShardId(), record.getSequenceNumber());
+ if (failedPerShard.get(kinesisMessageId.getShardId()).contains(new BigInteger(current.getSequenceNumber()))) {
+ failedandFetchedRecords.put(current, record);
+ shardIteratorPerFailedMessage.remove(current);
+ }
+ }
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
--- End diff --
lets make these warns for all the exceptions unless you think this might result lot of noise. Even than it will be helpful to print them given that these are critical errors.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1586
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72291464
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/Config.java ---
@@ -0,0 +1,166 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class Config implements Serializable {
+ // kinesis stream name to read from
+ private final String streamName;
+ // shard iterator type based on kinesis api - beginning of time, latest, at timestamp are only supported
+ private final ShardIteratorType shardIteratorType;
+ // implementation for converting a Kinesis record to a storm tuple
+ private final RecordToTupleMapper recordToTupleMapper;
+ // timestamp to be used for shardIteratorType AT_TIMESTAMP - can be null
+ private final Date timestamp;
+ // implementation for handling the failed messages retry logic
+ private final FailedMessageRetryHandler failedMessageRetryHandler;
+ // object capturing all zk related information for storing committed sequence numbers
+ private final ZkInfo zkInfo;
+ // object representing information on paramaters to use while connecting to kinesis using kinesis client
+ private final KinesisConnectionInfo kinesisConnectionInfo;
+ // this number represents the number of messages that are still not committed to zk. it will prevent the spout from emitting further.
+ // for e.g. if 1 failed and 2,3,4,5..... all have been acked by storm, they still cant be committed to zk because 1 is still in failed set. As a result
+ // the acked queue can infinitely grow without any of them being committed to zk. topology max pending does not help since from storm's view they are acked
+ private final Long maxUncommittedRecords;
+
+ public Config (String streamName, ShardIteratorType shardIteratorType, RecordToTupleMapper recordToTupleMapper, Date timestamp, FailedMessageRetryHandler
+ failedMessageRetryHandler, ZkInfo zkInfo, KinesisConnectionInfo kinesisConnectionInfo, Long maxUncommittedRecords) {
+ this.streamName = streamName;
+ this.shardIteratorType = shardIteratorType;
+ this.recordToTupleMapper = recordToTupleMapper;
+ this.timestamp = timestamp;
+ this.failedMessageRetryHandler = failedMessageRetryHandler;
+ this.zkInfo = zkInfo;
+ this.kinesisConnectionInfo = kinesisConnectionInfo;
+ this.maxUncommittedRecords = maxUncommittedRecords;
+ validate();
+ }
+
+ private void validate () {
+ if (streamName == null || streamName.length() < 1) {
+ throw new IllegalArgumentException("streamName is required and cannot be of length 0.");
+ }
+ if (shardIteratorType == null || shardIteratorType.equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER) || shardIteratorType.equals(ShardIteratorType
+ .AT_SEQUENCE_NUMBER)) {
+ throw new IllegalArgumentException("shardIteratorType has to be one of the " + ShardIteratorType.AT_TIMESTAMP + "," + ShardIteratorType.LATEST +
+ "," + ShardIteratorType.TRIM_HORIZON);
+ }
+ if (shardIteratorType.equals(ShardIteratorType.AT_TIMESTAMP) && timestamp == null) {
+ throw new IllegalArgumentException("timestamp must be provided if shardIteratorType is " + ShardIteratorType.AT_TIMESTAMP);
+ }
+ if (recordToTupleMapper == null) {
+ throw new IllegalArgumentException("recordToTupleMapper cannot be null");
+ }
+ if (failedMessageRetryHandler == null) {
+ throw new IllegalArgumentException("failedMessageRetryHandler cannot be null");
+ }
+ if (zkInfo == null) {
+ throw new IllegalArgumentException("zkInfo cannot be null");
+ }
+ if (kinesisConnectionInfo == null) {
+ throw new IllegalArgumentException("kinesisConnectionInfo cannot be null");
+ }
+ if (maxUncommittedRecords == null || maxUncommittedRecords < 1) {
+ throw new IllegalArgumentException("maxUncommittedRecords has to be a positive integer");
+ }
+ }
+
+ public String getStreamName() {
+ return streamName;
+ }
+
+ public ShardIteratorType getShardIteratorType() {
+ return shardIteratorType;
+ }
+
+ public RecordToTupleMapper getRecordToTupleMapper() {
+ return recordToTupleMapper;
+ }
+
+ public Date getTimestamp() {
+ return timestamp;
+ }
+
+ public FailedMessageRetryHandler getFailedMessageRetryHandler () {
+ return failedMessageRetryHandler;
+ }
+
+ public ZkInfo getZkInfo () {
+ return zkInfo;
+ }
+
+ public KinesisConnectionInfo getKinesisConnectionInfo () {
+ return kinesisConnectionInfo;
+ }
+
+ public Long getMaxUncommittedRecords () {
+ return maxUncommittedRecords;
+ }
+
+ @Override
+ public boolean equals(Object o) {
--- End diff --
We dont need equals and hashcode. I will remove it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72298716
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
--- End diff --
should this be configurable ? or probably a candidate for exponential back off . we can address this in follow-up jira
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72294716
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisConnectionInfo.java ---
@@ -0,0 +1,137 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.regions.Regions;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Serializable;
+import java.util.Arrays;
+
+public class KinesisConnectionInfo implements Serializable {
+ private final byte[] serializedKinesisCredsProvider;
+ private final byte[] serializedkinesisClientConfig;
+ private final Integer recordsLimit;
+ private final Regions region;
+
+ private transient AWSCredentialsProvider credentialsProvider;
+ private transient ClientConfiguration clientConfiguration;
+
+ /**
+ *
+ * @param credentialsProvider implementation to provide credentials to connect to kinesis
+ * @param clientConfiguration client configuration to pass to kinesis client
+ * @param region region to connect to
+ * @param recordsLimit max records to be fetched in a getRecords request to kinesis
+ */
+ public KinesisConnectionInfo (AWSCredentialsProvider credentialsProvider, ClientConfiguration clientConfiguration, Regions region, Integer recordsLimit) {
+ if (recordsLimit == null || recordsLimit <= 0) {
+ throw new IllegalArgumentException("recordsLimit has to be a positive integer");
+ }
+ if (region == null) {
+ throw new IllegalArgumentException("region cannot be null");
+ }
+ serializedKinesisCredsProvider = getKryoSerializedBytes(credentialsProvider);
+ serializedkinesisClientConfig = getKryoSerializedBytes(clientConfiguration);
+ this.recordsLimit = recordsLimit;
+ this.region = region;
+
+ this.credentialsProvider = null;
+ this.clientConfiguration = null;
+ }
+
+ public Integer getRecordsLimit() {
+ return recordsLimit;
+ }
+
+ public AWSCredentialsProvider getCredentialsProvider() {
+ if (credentialsProvider == null) {
+ credentialsProvider = (AWSCredentialsProvider) this.getKryoDeserializedObject(serializedKinesisCredsProvider);
+ }
+ return credentialsProvider;
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ if (clientConfiguration == null) {
+ clientConfiguration = (ClientConfiguration) this.getKryoDeserializedObject(serializedkinesisClientConfig);
+ }
+ return clientConfiguration;
+ }
+
+ public Regions getRegion() {
+ return region;
+ }
+
+ private byte[] getKryoSerializedBytes (final Object obj) {
+ final Kryo kryo = new Kryo();
+ final ByteArrayOutputStream os = new ByteArrayOutputStream();
+ final Output output = new Output(os);
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ kryo.writeClassAndObject(output, obj);
+ output.flush();
+ return os.toByteArray();
+ }
+
+ private Object getKryoDeserializedObject (final byte[] ser) {
+ final Kryo kryo = new Kryo();
+ final Input input = new Input(new ByteArrayInputStream(ser));
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ return kryo.readClassAndObject(input);
+ }
+
+ @Override
+ public boolean equals(Object o) {
--- End diff --
we probably can remove these equals, hash here as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72292144
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
--- End diff --
ExponentialBackoffRetrier has two constructors. The no args constructor will set the default values for these variables.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72279789
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
--- End diff --
should we have any defaults for these variables.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the issue:
https://github.com/apache/storm/pull/1586
@priyank5485 left couple of comments
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72305189
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
--- End diff --
I thought about it but if i try to reduce the return statements it will create more if else statements. If its not a big issue we can keep as is.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72291240
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/Config.java ---
@@ -0,0 +1,166 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class Config implements Serializable {
--- End diff --
Will change it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72279392
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
+ return false;
+ }
+ // if first failure add it to the count map
+ if (!failCounts.containsKey(messageId)) {
+ failCounts.put(messageId, 0L);
+ }
+ // increment the fail count as we started with 0
+ Long failCount = failCounts.get(messageId);
+ failCounts.put(messageId, ++failCount);
+ // if fail count is greater than maxRetries, discard or ack. for e.g. for maxRetries 3, 4 failures are allowed at maximum
+ if (failCount > maxRetries) {
+ LOG.debug("maxRetries reached so dropping " + messageId);
--- End diff --
can you make this as warn as it can be useful to see this in the logs by default
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72303034
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisConnectionInfo.java ---
@@ -0,0 +1,137 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.regions.Regions;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Serializable;
+import java.util.Arrays;
+
+public class KinesisConnectionInfo implements Serializable {
+ private final byte[] serializedKinesisCredsProvider;
+ private final byte[] serializedkinesisClientConfig;
+ private final Integer recordsLimit;
+ private final Regions region;
+
+ private transient AWSCredentialsProvider credentialsProvider;
+ private transient ClientConfiguration clientConfiguration;
+
+ /**
+ *
+ * @param credentialsProvider implementation to provide credentials to connect to kinesis
+ * @param clientConfiguration client configuration to pass to kinesis client
+ * @param region region to connect to
+ * @param recordsLimit max records to be fetched in a getRecords request to kinesis
+ */
+ public KinesisConnectionInfo (AWSCredentialsProvider credentialsProvider, ClientConfiguration clientConfiguration, Regions region, Integer recordsLimit) {
+ if (recordsLimit == null || recordsLimit <= 0) {
+ throw new IllegalArgumentException("recordsLimit has to be a positive integer");
+ }
+ if (region == null) {
+ throw new IllegalArgumentException("region cannot be null");
+ }
+ serializedKinesisCredsProvider = getKryoSerializedBytes(credentialsProvider);
+ serializedkinesisClientConfig = getKryoSerializedBytes(clientConfiguration);
+ this.recordsLimit = recordsLimit;
+ this.region = region;
+
+ this.credentialsProvider = null;
+ this.clientConfiguration = null;
+ }
+
+ public Integer getRecordsLimit() {
+ return recordsLimit;
+ }
+
+ public AWSCredentialsProvider getCredentialsProvider() {
+ if (credentialsProvider == null) {
+ credentialsProvider = (AWSCredentialsProvider) this.getKryoDeserializedObject(serializedKinesisCredsProvider);
+ }
+ return credentialsProvider;
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ if (clientConfiguration == null) {
+ clientConfiguration = (ClientConfiguration) this.getKryoDeserializedObject(serializedkinesisClientConfig);
+ }
+ return clientConfiguration;
+ }
+
+ public Regions getRegion() {
+ return region;
+ }
+
+ private byte[] getKryoSerializedBytes (final Object obj) {
+ final Kryo kryo = new Kryo();
+ final ByteArrayOutputStream os = new ByteArrayOutputStream();
+ final Output output = new Output(os);
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ kryo.writeClassAndObject(output, obj);
+ output.flush();
+ return os.toByteArray();
+ }
+
+ private Object getKryoDeserializedObject (final byte[] ser) {
+ final Kryo kryo = new Kryo();
+ final Input input = new Input(new ByteArrayInputStream(ser));
+ kryo.setInstantiatorStrategy(new StdInstantiatorStrategy());
+ return kryo.readClassAndObject(input);
+ }
+
+ @Override
+ public boolean equals(Object o) {
--- End diff --
Will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the issue:
https://github.com/apache/storm/pull/1586
@priyank5485 looks great!. Mostly nit-picks. I'll go over the retry handle one more time and add any comments if needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72299262
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ // add all fetched records to the set of failed records if they are present in failed set
+ for (Record record: records) {
+ KinesisMessageId current = new KinesisMessageId(kinesisMessageId.getStreamName(), kinesisMessageId.getShardId(), record.getSequenceNumber());
+ if (failedPerShard.get(kinesisMessageId.getShardId()).contains(new BigInteger(current.getSequenceNumber()))) {
+ failedandFetchedRecords.put(current, record);
+ shardIteratorPerFailedMessage.remove(current);
+ }
+ }
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
+ } catch (ExpiredIteratorException ex) {
+ LOG.debug("shardIterator for failedRecord " + kinesisMessageId + " has expired. Refreshing shardIterator");
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ } catch (ProvisionedThroughputExceededException pe) {
+ try {
+ LOG.debug("ProvisionedThroughputExceededException occured. Check your limits. Sleeping for 1 second.", pe);
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.debug("Thread interrupted exception", e);
+ }
+ }
+ }
+
+ private void fetchNewRecords () {
+ for (Map.Entry<String, LinkedList<Record>> entry : toEmitPerShard.entrySet()) {
+ String shardId = entry.getKey();
+ try {
+ String shardIterator = shardIteratorPerShard.get(shardId);
+ LOG.debug("Fetching new records for shard id :" + shardId + " using shardIterator " + shardIterator + " after sequence number " +
+ fetchedSequenceNumberPerShard.get(shardId));
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchNewRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerShard.put(shardId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ entry.getValue().addAll(records);
+ fetchedSequenceNumberPerShard.put(shardId, records.get(records.size() - 1).getSequenceNumber());
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
+ } catch (ExpiredIteratorException ex) {
+ LOG.debug("shardIterator for shardId " + shardId + " has expired. Refreshing shardIterator");
+ refreshShardIteratorForNewRecords(shardId);
+ } catch (ProvisionedThroughputExceededException pe) {
+ try {
+ LOG.debug("ProvisionedThroughputExceededException occured. Check your limits. Sleeping for 1 second.", pe);
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.debug("Thread interrupted exception", e);
+ }
+ }
+ }
+ }
+
+ private GetRecordsResult fetchRecords (String shardIterator) {
+ List<Record> records = new ArrayList<>();
+ GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
+ getRecordsRequest.setShardIterator(shardIterator);
+ getRecordsRequest.setLimit(config.getKinesisConnectionInfo().getRecordsLimit());
+ GetRecordsResult getRecordsResult = kinesisClient.getRecords(getRecordsRequest);
+ return getRecordsResult;
+ }
+
+ private List<Shard> getShards () {
+ DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
+ describeStreamRequest.setStreamName(config.getStreamName());
+ List<Shard> shards = new ArrayList<>();
+ String exclusiveStartShardId = null;
+ do {
+ describeStreamRequest.setExclusiveStartShardId(exclusiveStartShardId);
+ DescribeStreamResult describeStreamResult = kinesisClient.describeStream(describeStreamRequest);
+ shards.addAll(describeStreamResult.getStreamDescription().getShards());
+ if (describeStreamResult.getStreamDescription().getHasMoreShards() && shards.size() > 0) {
+ exclusiveStartShardId = shards.get(shards.size() - 1).getShardId();
+ } else {
+ exclusiveStartShardId = null;
+ }
+ } while ( exclusiveStartShardId != null );
+ LOG.info("Number of shards for stream " + config.getStreamName() + " are " + shards.size());
+ return shards;
+ }
+
+ private void emitNewRecord (SpoutOutputCollector collector) {
+ for (Map.Entry<String, LinkedList<Record>> entry: toEmitPerShard.entrySet()) {
+ String shardId = entry.getKey();
+ LinkedList<Record> listOfRecords = entry.getValue();
+ Record record;
+ while ((record = listOfRecords.pollFirst()) != null) {
+ KinesisMessageId kinesisMessageId = new KinesisMessageId(config.getStreamName(), shardId, record.getSequenceNumber());
+ if (emitRecord(collector, record, kinesisMessageId)) {
+ return;
+ }
+ }
+ }
+ }
+
+ private boolean emitFailedRecord (SpoutOutputCollector collector, KinesisMessageId kinesisMessageId) {
+ if (!failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ return false;
+ }
+ return emitRecord(collector, failedandFetchedRecords.get(kinesisMessageId), kinesisMessageId);
+ }
+
+ private boolean emitRecord (SpoutOutputCollector collector, Record record, KinesisMessageId kinesisMessageId) {
+ boolean result = false;
+ List<Object> tuple = config.getRecordToTupleMapper().getTuple(record);
+ // if a record is returned put the sequence number in the emittedPerShard to tie back with ack or fail
+ if (tuple != null && tuple.size() > 0) {
+ collector.emit(tuple, kinesisMessageId);
+ if (!emittedPerShard.containsKey(kinesisMessageId.getShardId())) {
+ emittedPerShard.put(kinesisMessageId.getShardId(), new TreeSet<BigInteger>());
+ }
+ emittedPerShard.get(kinesisMessageId.getShardId()).add(new BigInteger(record.getSequenceNumber()));
+ result = true;
+ } else {
+ // ack to not process the record again on restart and move on to next message
+ LOG.debug("Record " + record + " did not return a tuple to emit. Hence acking it");
+ ack(kinesisMessageId);
+ }
+ return result;
+ }
+
+ private boolean shouldCommit () {
+ return (System.currentTimeMillis() - lastCommitTime >= config.getZkInfo().getCommitIntervalMs());
+ }
+
+ private void initializeFetchedSequenceNumbers () {
+ for (String shardId : toEmitPerShard.keySet()) {
+ Map<Object, Object> state = readState(getZkPath(shardId));
+ // if state found for this shard in zk, then set the sequence number in fetchedSequenceNumber
+ if (state != null) {
+ Object committedSequenceNumber = state.get("committedSequenceNumber");
+ LOG.info("State read is committedSequenceNumber: " + committedSequenceNumber + " shardId:" + shardId);
+ if (committedSequenceNumber != null) {
+ fetchedSequenceNumberPerShard.put(shardId, (String) committedSequenceNumber);
+ }
+ }
+ }
+ }
+
+ private void refreshShardIteratorsForNewRecords () {
+ for (String shardId: toEmitPerShard.keySet()) {
+ refreshShardIteratorForNewRecords(shardId);
+ }
+ }
+
+ private void refreshShardIteratorForNewRecords (String shardId) {
+ String shardIterator = null;
+ String lastFetchedSequenceNumber = fetchedSequenceNumberPerShard.get(shardId);
+ ShardIteratorType shardIteratorType = (lastFetchedSequenceNumber == null ? config.getShardIteratorType() : ShardIteratorType
+ .AFTER_SEQUENCE_NUMBER);
+ // Set the shard iterator for last fetched sequence number to start from correct position in shard
+ shardIterator = this.getShardIterator(shardId, shardIteratorType, lastFetchedSequenceNumber, config.getTimestamp());
+ if (shardIterator != null && !shardIterator.isEmpty()) {
+ LOG.debug("Refreshing shard iterator for new records for shardId " + shardId + " with shardIterator " + shardIterator);
+ shardIteratorPerShard.put(shardId, shardIterator);
+ }
+ }
+
+ private void refreshShardIteratorForFailedRecord (KinesisMessageId kinesisMessageId) {
+ String shardIterator = null;
+ // Set the shard iterator for last fetched sequence number to start from correct position in shard
+ shardIterator = this.getShardIterator(kinesisMessageId.getShardId(), ShardIteratorType.AT_SEQUENCE_NUMBER, kinesisMessageId.getSequenceNumber(), null);
+ if (shardIterator != null && !shardIterator.isEmpty()) {
+ LOG.debug("Refreshing shard iterator for failed records for message " + kinesisMessageId + " with shardIterator " + shardIterator);
+ shardIteratorPerFailedMessage.put(kinesisMessageId, shardIterator);
+ }
+ }
+
+ private String getShardIterator (String shardId, ShardIteratorType shardIteratorType, String sequenceNumber, Date timestamp) {
+ String shardIterator = "";
+ try {
+ GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
+ getShardIteratorRequest.setStreamName(config.getStreamName());
+ getShardIteratorRequest.setShardId(shardId);
+ getShardIteratorRequest.setShardIteratorType(shardIteratorType);
+ if (shardIteratorType.equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER) || shardIteratorType.equals(ShardIteratorType.AT_SEQUENCE_NUMBER)) {
+ getShardIteratorRequest.setStartingSequenceNumber(sequenceNumber);
+ } else if (shardIteratorType.equals(ShardIteratorType.AT_TIMESTAMP)) {
+ getShardIteratorRequest.setTimestamp(timestamp);
+ }
+ GetShardIteratorResult getShardIteratorResult = kinesisClient.getShardIterator(getShardIteratorRequest);
+ if (getShardIteratorResult != null) {
+ shardIterator = getShardIteratorResult.getShardIterator();
+ }
+ } catch (Exception e) {
+ LOG.debug("Exception occured while getting shardIterator for shard " + shardId + " shardIteratorType " + shardIteratorType + " sequence number " +
+ sequenceNumber + " timestamp " + timestamp, e);
+ }
+ LOG.debug("Returning shardIterator " + shardIterator + " for shardId " + shardId + " shardIteratorType " + shardIteratorType + " sequenceNumber " +
+ sequenceNumber + " timestamp" + timestamp);
+ return shardIterator;
+ }
+
+ private Long getUncommittedRecordsCount () {
+ Long result = 0L;
+ for (Map.Entry<String, TreeSet<BigInteger>> emitted: emittedPerShard.entrySet()) {
+ result += emitted.getValue().size();
+ }
+ for (Map.Entry<String, TreeSet<BigInteger>> acked: ackedPerShard.entrySet()) {
+ result += acked.getValue().size();
+ }
+ for (Map.Entry<String, TreeSet<BigInteger>> failed: failedPerShard.entrySet()) {
+ result += failed.getValue().size();
+ }
+ LOG.debug("Returning uncommittedRecordsCount as " + result);
+ return result;
+ }
+
+ private boolean shouldFetchNewRecords () {
+ // check to see if any shard has already fetched records waiting to be emitted, in which case dont fetch more
+ boolean fetchRecords = true;
+ for (Map.Entry<String, LinkedList<Record>> entry: toEmitPerShard.entrySet()) {
+ if (!entry.getValue().isEmpty()) {
+ fetchRecords = false;
+ break;
+ }
+ }
+ return fetchRecords;
+ }
+
+ private void initializeCurator () {
+ ZkInfo zkInfo = config.getZkInfo();
+ curatorFramework = CuratorFrameworkFactory.newClient(zkInfo.getZkUrl(), zkInfo.getSessionTimeoutMs(), zkInfo.getConnectionTimeoutMs(), new
+ RetryNTimes(zkInfo.getRetryAttempts(), zkInfo.getRetryIntervalMs()));
+ curatorFramework.start();
+ }
+
+ private void initializeKinesisClient () {
+ kinesisClient = new AmazonKinesisClient(config.getKinesisConnectionInfo().getCredentialsProvider(), config.getKinesisConnectionInfo().getClientConfiguration());
+ kinesisClient.setRegion(Region.getRegion(config.getKinesisConnectionInfo().getRegion()));
+ }
+
+ private void shutdownCurator () {
+ curatorFramework.close();
+ }
+
+ private void shutdownKinesisClient () {
+ kinesisClient.shutdown();
+ }
+
+}
--- End diff --
This is big file thats doing lot of things. We should some of these methods into Utils class or something like zk related methods into its own class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72279114
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/Config.java ---
@@ -0,0 +1,166 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class Config implements Serializable {
+ // kinesis stream name to read from
+ private final String streamName;
+ // shard iterator type based on kinesis api - beginning of time, latest, at timestamp are only supported
+ private final ShardIteratorType shardIteratorType;
+ // implementation for converting a Kinesis record to a storm tuple
+ private final RecordToTupleMapper recordToTupleMapper;
+ // timestamp to be used for shardIteratorType AT_TIMESTAMP - can be null
+ private final Date timestamp;
+ // implementation for handling the failed messages retry logic
+ private final FailedMessageRetryHandler failedMessageRetryHandler;
+ // object capturing all zk related information for storing committed sequence numbers
+ private final ZkInfo zkInfo;
+ // object representing information on paramaters to use while connecting to kinesis using kinesis client
+ private final KinesisConnectionInfo kinesisConnectionInfo;
+ // this number represents the number of messages that are still not committed to zk. it will prevent the spout from emitting further.
+ // for e.g. if 1 failed and 2,3,4,5..... all have been acked by storm, they still cant be committed to zk because 1 is still in failed set. As a result
+ // the acked queue can infinitely grow without any of them being committed to zk. topology max pending does not help since from storm's view they are acked
+ private final Long maxUncommittedRecords;
+
+ public Config (String streamName, ShardIteratorType shardIteratorType, RecordToTupleMapper recordToTupleMapper, Date timestamp, FailedMessageRetryHandler
+ failedMessageRetryHandler, ZkInfo zkInfo, KinesisConnectionInfo kinesisConnectionInfo, Long maxUncommittedRecords) {
+ this.streamName = streamName;
+ this.shardIteratorType = shardIteratorType;
+ this.recordToTupleMapper = recordToTupleMapper;
+ this.timestamp = timestamp;
+ this.failedMessageRetryHandler = failedMessageRetryHandler;
+ this.zkInfo = zkInfo;
+ this.kinesisConnectionInfo = kinesisConnectionInfo;
+ this.maxUncommittedRecords = maxUncommittedRecords;
+ validate();
+ }
+
+ private void validate () {
+ if (streamName == null || streamName.length() < 1) {
+ throw new IllegalArgumentException("streamName is required and cannot be of length 0.");
+ }
+ if (shardIteratorType == null || shardIteratorType.equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER) || shardIteratorType.equals(ShardIteratorType
+ .AT_SEQUENCE_NUMBER)) {
+ throw new IllegalArgumentException("shardIteratorType has to be one of the " + ShardIteratorType.AT_TIMESTAMP + "," + ShardIteratorType.LATEST +
+ "," + ShardIteratorType.TRIM_HORIZON);
+ }
+ if (shardIteratorType.equals(ShardIteratorType.AT_TIMESTAMP) && timestamp == null) {
+ throw new IllegalArgumentException("timestamp must be provided if shardIteratorType is " + ShardIteratorType.AT_TIMESTAMP);
+ }
+ if (recordToTupleMapper == null) {
+ throw new IllegalArgumentException("recordToTupleMapper cannot be null");
+ }
+ if (failedMessageRetryHandler == null) {
+ throw new IllegalArgumentException("failedMessageRetryHandler cannot be null");
+ }
+ if (zkInfo == null) {
+ throw new IllegalArgumentException("zkInfo cannot be null");
+ }
+ if (kinesisConnectionInfo == null) {
+ throw new IllegalArgumentException("kinesisConnectionInfo cannot be null");
+ }
+ if (maxUncommittedRecords == null || maxUncommittedRecords < 1) {
+ throw new IllegalArgumentException("maxUncommittedRecords has to be a positive integer");
+ }
+ }
+
+ public String getStreamName() {
+ return streamName;
+ }
+
+ public ShardIteratorType getShardIteratorType() {
+ return shardIteratorType;
+ }
+
+ public RecordToTupleMapper getRecordToTupleMapper() {
+ return recordToTupleMapper;
+ }
+
+ public Date getTimestamp() {
+ return timestamp;
+ }
+
+ public FailedMessageRetryHandler getFailedMessageRetryHandler () {
+ return failedMessageRetryHandler;
+ }
+
+ public ZkInfo getZkInfo () {
+ return zkInfo;
+ }
+
+ public KinesisConnectionInfo getKinesisConnectionInfo () {
+ return kinesisConnectionInfo;
+ }
+
+ public Long getMaxUncommittedRecords () {
+ return maxUncommittedRecords;
+ }
+
+ @Override
+ public boolean equals(Object o) {
--- End diff --
Do we need equals and hash code . Not sure where do we do comparisons of config object.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r73043179
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,449 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // object handling zk interaction
+ private transient ZKConnection zkConnection;
+ // object handling interaction with kinesis
+ private transient KinesisConnection kinesisConnection;
+ // Kinesis Spout KinesisConfig object
+ private transient final KinesisConfig kinesisConfig;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+
+ KinesisRecordsManager (KinesisConfig kinesisConfig) {
+ this.kinesisConfig = kinesisConfig;
+ this.zkConnection = new ZKConnection(kinesisConfig.getZkInfo());
+ this.kinesisConnection = new KinesisConnection(kinesisConfig.getKinesisConnectionInfo());
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ kinesisConnection.initialize();
+ zkConnection.initialize();
+ List<Shard> shards = kinesisConnection.getShardsForStream(kinesisConfig.getStreamName());
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = kinesisConfig.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ kinesisConfig.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.warn("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.warn("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < kinesisConfig.getMaxUncommittedRecords())) {
+ LOG.warn("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.warn("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // if an ack is received for a message then add it to the ackedPerShard TreeSet. TreeSet because while committing we need to figure out what is the
+ // highest sequence number that can be committed for this shard
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ // if the acked message was in emittedPerShard that means we need to remove it from the emittedPerShard(which keeps track of in flight tuples)
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ // an acked message should not be in failed since if it fails and gets re-emitted it moves to emittedPerShard from failedPerShard. Defensive coding.
+ // Remove it from failedPerShard anyway
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ // if an ack is for a message that failed once at least and was re-emitted then the record itself will be in failedAndFetchedRecords. We use that to
+ // determine if the FailedMessageRetryHandler needs to be told about it and then remove the record itself to clean up memory
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ kinesisConfig.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (kinesisConfig.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // We have three mutually disjoint treesets per shard at any given time to keep track of what sequence number can be committed to zookeeper.
+ // emittedPerShard, ackedPerShard and failedPerShard. Any record starts by entering emittedPerShard. On ack it moves from emittedPerShard to
+ // ackedPerShard and on fail if retry service tells us to retry then it moves from emittedPerShard to failedPerShard. The failed records will move from
+ // failedPerShard to emittedPerShard when the failed record is emitted again as a retry.
+ // Logic for deciding what sequence number to commit is find the highest sequence number from ackedPerShard called X such that there is no sequence
+ // number Y in emittedPerShard or failedPerShard that satisfies X > Y. For e.g. if ackedPerShard is 1,4,5, emittedPerShard is 2,6 and
+ // failedPerShard is 3,7 then we can only commit 1 and not 4 because 2 is still pending and 3 has failed
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ zkConnection.commitState(kinesisConfig.getStreamName(), shardId, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ kinesisConnection.initialize();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ kinesisConnection.shutdown();
+ }
+
+ void close () {
+ commit();
+ kinesisConnection.shutdown();
+ zkConnection.shutdown();
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = kinesisConnection.fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.warn("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
--- End diff --
Sorry missed these earlier. Any reason for it to sleep exactly one sec or should it be configurable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72281666
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
--- End diff --
are we going to store the same message in retryMessageSet and retryTimes?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72291774
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
+ return false;
+ }
+ // if first failure add it to the count map
+ if (!failCounts.containsKey(messageId)) {
+ failCounts.put(messageId, 0L);
+ }
+ // increment the fail count as we started with 0
+ Long failCount = failCounts.get(messageId);
+ failCounts.put(messageId, ++failCount);
+ // if fail count is greater than maxRetries, discard or ack. for e.g. for maxRetries 3, 4 failures are allowed at maximum
+ if (failCount > maxRetries) {
+ LOG.debug("maxRetries reached so dropping " + messageId);
+ failCounts.remove(messageId);
+ return false;
+ }
+ // if reached so far, add it to the set of messages waiting to be retried with next retry time based on how many times it failed
+ retryTimes.put(messageId, getRetryTime(failCount));
+ retryMessageSet.add(messageId);
+ LOG.debug("Scheduled " + messageId + " for retry at " + retryTimes.get(messageId) + " and retry attempt " + failCount);
--- End diff --
I will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72279515
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
--- End diff --
warn here as well
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72300622
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ZkInfo.java ---
@@ -0,0 +1,153 @@
+/**
+ * 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.kinesis.spout;
+
+import java.io.Serializable;
+
+public class ZkInfo implements Serializable {
+ // comma separated list of zk connect strings to connect to zookeeper e.g. localhost:2181
+ private final String zkUrl;
+ // zk node under which to commit the sequence number of messages. e.g. /committed_sequence_numbers
+ private final String zkNode;
+ // zk session timeout in milliseconds
+ private final Integer sessionTimeoutMs;
+ // zk connection timeout in milliseconds
+ private final Integer connectionTimeoutMs;
+ // interval at which to commit offsets to zk in milliseconds
+ private final Long commitIntervalMs;
+ // number of retry attempts for zk
+ private final Integer retryAttempts;
+ // time to sleep between retries in milliseconds
+ private final Integer retryIntervalMs;
+
+ /**
+ * Default constructor that uses defaults for a local setup
+ */
+ public ZkInfo () {
+ this("localhost:2181", "/kinesisOffsets", 20000, 15000, 10000L, 3, 2000);
+ }
+
+ public ZkInfo (String zkUrl, String zkNode, Integer sessionTimeoutMs, Integer connectionTimeoutMs, Long commitIntervalMs, Integer retryAttempts, Integer
+ retryIntervalMs) {
+ this.zkUrl = zkUrl;
+ this.zkNode = zkNode;
+ this.sessionTimeoutMs = sessionTimeoutMs;
+ this.connectionTimeoutMs = connectionTimeoutMs;
+ this.commitIntervalMs = commitIntervalMs;
+ this.retryAttempts = retryAttempts;
+ this.retryIntervalMs = retryIntervalMs;
+ validate();
+ }
+
+ public String getZkUrl() {
+ return zkUrl;
+ }
+
+ public String getZkNode() {
+ return zkNode;
+ }
+
+ public Integer getSessionTimeoutMs() {
+ return sessionTimeoutMs;
+ }
+
+ public Integer getConnectionTimeoutMs() {
+ return connectionTimeoutMs;
+ }
+
+ public Long getCommitIntervalMs() {
+ return commitIntervalMs;
+ }
+
+ public Integer getRetryAttempts() {
+ return retryAttempts;
+ }
+
+ public Integer getRetryIntervalMs() {
+ return retryIntervalMs;
+ }
+
+ private void validate () {
+
+ if (zkUrl == null || zkUrl.length() < 1) {
+ throw new IllegalArgumentException("zkUrl must be specified to connect to zookeeper");
+ }
+ if (zkNode == null || zkNode.length() < 1) {
+ throw new IllegalArgumentException("zkNode must be specified");
+ }
+ checkPositive(sessionTimeoutMs, "sessionTimeoutMs");
+ checkPositive(connectionTimeoutMs, "connectionTimeoutMs");
+ checkPositive(commitIntervalMs, "commitIntervalMs");
+ checkPositive(retryAttempts, "retryAttempts");
+ checkPositive(retryIntervalMs, "retryIntervalMs");
+ }
+
+ private void checkPositive (Integer argument, String name) {
+ if (argument == null && argument <= 0) {
+ throw new IllegalArgumentException(name + " must be positive");
+ }
+ }
+ private void checkPositive (Long argument, String name) {
+ if (argument == null && argument <= 0) {
+ throw new IllegalArgumentException(name + " must be positive");
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "ZkInfo{" +
+ "zkUrl='" + zkUrl + '\'' +
+ ", zkNode='" + zkNode + '\'' +
+ ", sessionTimeoutMs=" + sessionTimeoutMs +
+ ", connectionTimeoutMs=" + connectionTimeoutMs +
+ ", commitIntervalMs=" + commitIntervalMs +
+ ", retryAttempts=" + retryAttempts +
+ ", retryIntervalMs=" + retryIntervalMs +
+ '}';
+ }
+
+ @Override
--- End diff --
remove these as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72305349
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
--- End diff --
I documented the logic in the commit method since thats where it writes to zookeeper. Let me know if that is not enough or if i should move it to somewhere else.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72291939
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
--- End diff --
I will change it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72921532
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ // add all fetched records to the set of failed records if they are present in failed set
+ for (Record record: records) {
+ KinesisMessageId current = new KinesisMessageId(kinesisMessageId.getStreamName(), kinesisMessageId.getShardId(), record.getSequenceNumber());
+ if (failedPerShard.get(kinesisMessageId.getShardId()).contains(new BigInteger(current.getSequenceNumber()))) {
+ failedandFetchedRecords.put(current, record);
+ shardIteratorPerFailedMessage.remove(current);
+ }
+ }
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
+ } catch (ExpiredIteratorException ex) {
+ LOG.debug("shardIterator for failedRecord " + kinesisMessageId + " has expired. Refreshing shardIterator");
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ } catch (ProvisionedThroughputExceededException pe) {
+ try {
+ LOG.debug("ProvisionedThroughputExceededException occured. Check your limits. Sleeping for 1 second.", pe);
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.debug("Thread interrupted exception", e);
+ }
+ }
+ }
+
+ private void fetchNewRecords () {
+ for (Map.Entry<String, LinkedList<Record>> entry : toEmitPerShard.entrySet()) {
+ String shardId = entry.getKey();
+ try {
+ String shardIterator = shardIteratorPerShard.get(shardId);
+ LOG.debug("Fetching new records for shard id :" + shardId + " using shardIterator " + shardIterator + " after sequence number " +
+ fetchedSequenceNumberPerShard.get(shardId));
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchNewRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerShard.put(shardId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ entry.getValue().addAll(records);
+ fetchedSequenceNumberPerShard.put(shardId, records.get(records.size() - 1).getSequenceNumber());
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
+ } catch (ExpiredIteratorException ex) {
+ LOG.debug("shardIterator for shardId " + shardId + " has expired. Refreshing shardIterator");
+ refreshShardIteratorForNewRecords(shardId);
+ } catch (ProvisionedThroughputExceededException pe) {
+ try {
+ LOG.debug("ProvisionedThroughputExceededException occured. Check your limits. Sleeping for 1 second.", pe);
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ LOG.debug("Thread interrupted exception", e);
+ }
+ }
+ }
+ }
+
+ private GetRecordsResult fetchRecords (String shardIterator) {
+ List<Record> records = new ArrayList<>();
+ GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
+ getRecordsRequest.setShardIterator(shardIterator);
+ getRecordsRequest.setLimit(config.getKinesisConnectionInfo().getRecordsLimit());
+ GetRecordsResult getRecordsResult = kinesisClient.getRecords(getRecordsRequest);
+ return getRecordsResult;
+ }
+
+ private List<Shard> getShards () {
+ DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
+ describeStreamRequest.setStreamName(config.getStreamName());
+ List<Shard> shards = new ArrayList<>();
+ String exclusiveStartShardId = null;
+ do {
+ describeStreamRequest.setExclusiveStartShardId(exclusiveStartShardId);
+ DescribeStreamResult describeStreamResult = kinesisClient.describeStream(describeStreamRequest);
+ shards.addAll(describeStreamResult.getStreamDescription().getShards());
+ if (describeStreamResult.getStreamDescription().getHasMoreShards() && shards.size() > 0) {
+ exclusiveStartShardId = shards.get(shards.size() - 1).getShardId();
+ } else {
+ exclusiveStartShardId = null;
+ }
+ } while ( exclusiveStartShardId != null );
+ LOG.info("Number of shards for stream " + config.getStreamName() + " are " + shards.size());
+ return shards;
+ }
+
+ private void emitNewRecord (SpoutOutputCollector collector) {
+ for (Map.Entry<String, LinkedList<Record>> entry: toEmitPerShard.entrySet()) {
+ String shardId = entry.getKey();
+ LinkedList<Record> listOfRecords = entry.getValue();
+ Record record;
+ while ((record = listOfRecords.pollFirst()) != null) {
+ KinesisMessageId kinesisMessageId = new KinesisMessageId(config.getStreamName(), shardId, record.getSequenceNumber());
+ if (emitRecord(collector, record, kinesisMessageId)) {
+ return;
+ }
+ }
+ }
+ }
+
+ private boolean emitFailedRecord (SpoutOutputCollector collector, KinesisMessageId kinesisMessageId) {
+ if (!failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ return false;
+ }
+ return emitRecord(collector, failedandFetchedRecords.get(kinesisMessageId), kinesisMessageId);
+ }
+
+ private boolean emitRecord (SpoutOutputCollector collector, Record record, KinesisMessageId kinesisMessageId) {
+ boolean result = false;
+ List<Object> tuple = config.getRecordToTupleMapper().getTuple(record);
+ // if a record is returned put the sequence number in the emittedPerShard to tie back with ack or fail
+ if (tuple != null && tuple.size() > 0) {
+ collector.emit(tuple, kinesisMessageId);
+ if (!emittedPerShard.containsKey(kinesisMessageId.getShardId())) {
+ emittedPerShard.put(kinesisMessageId.getShardId(), new TreeSet<BigInteger>());
+ }
+ emittedPerShard.get(kinesisMessageId.getShardId()).add(new BigInteger(record.getSequenceNumber()));
+ result = true;
+ } else {
+ // ack to not process the record again on restart and move on to next message
+ LOG.debug("Record " + record + " did not return a tuple to emit. Hence acking it");
+ ack(kinesisMessageId);
+ }
+ return result;
+ }
+
+ private boolean shouldCommit () {
+ return (System.currentTimeMillis() - lastCommitTime >= config.getZkInfo().getCommitIntervalMs());
+ }
+
+ private void initializeFetchedSequenceNumbers () {
+ for (String shardId : toEmitPerShard.keySet()) {
+ Map<Object, Object> state = readState(getZkPath(shardId));
+ // if state found for this shard in zk, then set the sequence number in fetchedSequenceNumber
+ if (state != null) {
+ Object committedSequenceNumber = state.get("committedSequenceNumber");
+ LOG.info("State read is committedSequenceNumber: " + committedSequenceNumber + " shardId:" + shardId);
+ if (committedSequenceNumber != null) {
+ fetchedSequenceNumberPerShard.put(shardId, (String) committedSequenceNumber);
+ }
+ }
+ }
+ }
+
+ private void refreshShardIteratorsForNewRecords () {
+ for (String shardId: toEmitPerShard.keySet()) {
+ refreshShardIteratorForNewRecords(shardId);
+ }
+ }
+
+ private void refreshShardIteratorForNewRecords (String shardId) {
+ String shardIterator = null;
+ String lastFetchedSequenceNumber = fetchedSequenceNumberPerShard.get(shardId);
+ ShardIteratorType shardIteratorType = (lastFetchedSequenceNumber == null ? config.getShardIteratorType() : ShardIteratorType
+ .AFTER_SEQUENCE_NUMBER);
+ // Set the shard iterator for last fetched sequence number to start from correct position in shard
+ shardIterator = this.getShardIterator(shardId, shardIteratorType, lastFetchedSequenceNumber, config.getTimestamp());
+ if (shardIterator != null && !shardIterator.isEmpty()) {
+ LOG.debug("Refreshing shard iterator for new records for shardId " + shardId + " with shardIterator " + shardIterator);
+ shardIteratorPerShard.put(shardId, shardIterator);
+ }
+ }
+
+ private void refreshShardIteratorForFailedRecord (KinesisMessageId kinesisMessageId) {
+ String shardIterator = null;
+ // Set the shard iterator for last fetched sequence number to start from correct position in shard
+ shardIterator = this.getShardIterator(kinesisMessageId.getShardId(), ShardIteratorType.AT_SEQUENCE_NUMBER, kinesisMessageId.getSequenceNumber(), null);
+ if (shardIterator != null && !shardIterator.isEmpty()) {
+ LOG.debug("Refreshing shard iterator for failed records for message " + kinesisMessageId + " with shardIterator " + shardIterator);
+ shardIteratorPerFailedMessage.put(kinesisMessageId, shardIterator);
+ }
+ }
+
+ private String getShardIterator (String shardId, ShardIteratorType shardIteratorType, String sequenceNumber, Date timestamp) {
+ String shardIterator = "";
+ try {
+ GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
+ getShardIteratorRequest.setStreamName(config.getStreamName());
+ getShardIteratorRequest.setShardId(shardId);
+ getShardIteratorRequest.setShardIteratorType(shardIteratorType);
+ if (shardIteratorType.equals(ShardIteratorType.AFTER_SEQUENCE_NUMBER) || shardIteratorType.equals(ShardIteratorType.AT_SEQUENCE_NUMBER)) {
+ getShardIteratorRequest.setStartingSequenceNumber(sequenceNumber);
+ } else if (shardIteratorType.equals(ShardIteratorType.AT_TIMESTAMP)) {
+ getShardIteratorRequest.setTimestamp(timestamp);
+ }
+ GetShardIteratorResult getShardIteratorResult = kinesisClient.getShardIterator(getShardIteratorRequest);
+ if (getShardIteratorResult != null) {
+ shardIterator = getShardIteratorResult.getShardIterator();
+ }
+ } catch (Exception e) {
+ LOG.debug("Exception occured while getting shardIterator for shard " + shardId + " shardIteratorType " + shardIteratorType + " sequence number " +
+ sequenceNumber + " timestamp " + timestamp, e);
+ }
+ LOG.debug("Returning shardIterator " + shardIterator + " for shardId " + shardId + " shardIteratorType " + shardIteratorType + " sequenceNumber " +
+ sequenceNumber + " timestamp" + timestamp);
+ return shardIterator;
+ }
+
+ private Long getUncommittedRecordsCount () {
+ Long result = 0L;
+ for (Map.Entry<String, TreeSet<BigInteger>> emitted: emittedPerShard.entrySet()) {
+ result += emitted.getValue().size();
+ }
+ for (Map.Entry<String, TreeSet<BigInteger>> acked: ackedPerShard.entrySet()) {
+ result += acked.getValue().size();
+ }
+ for (Map.Entry<String, TreeSet<BigInteger>> failed: failedPerShard.entrySet()) {
+ result += failed.getValue().size();
+ }
+ LOG.debug("Returning uncommittedRecordsCount as " + result);
+ return result;
+ }
+
+ private boolean shouldFetchNewRecords () {
+ // check to see if any shard has already fetched records waiting to be emitted, in which case dont fetch more
+ boolean fetchRecords = true;
+ for (Map.Entry<String, LinkedList<Record>> entry: toEmitPerShard.entrySet()) {
+ if (!entry.getValue().isEmpty()) {
+ fetchRecords = false;
+ break;
+ }
+ }
+ return fetchRecords;
+ }
+
+ private void initializeCurator () {
+ ZkInfo zkInfo = config.getZkInfo();
+ curatorFramework = CuratorFrameworkFactory.newClient(zkInfo.getZkUrl(), zkInfo.getSessionTimeoutMs(), zkInfo.getConnectionTimeoutMs(), new
+ RetryNTimes(zkInfo.getRetryAttempts(), zkInfo.getRetryIntervalMs()));
+ curatorFramework.start();
+ }
+
+ private void initializeKinesisClient () {
+ kinesisClient = new AmazonKinesisClient(config.getKinesisConnectionInfo().getCredentialsProvider(), config.getKinesisConnectionInfo().getClientConfiguration());
+ kinesisClient.setRegion(Region.getRegion(config.getKinesisConnectionInfo().getRegion()));
+ }
+
+ private void shutdownCurator () {
+ curatorFramework.close();
+ }
+
+ private void shutdownKinesisClient () {
+ kinesisClient.shutdown();
+ }
+
+}
--- End diff --
@priyank5485 do you think its possible to move some of these methods into its Utils file or some sorts.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72306731
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ZkInfo.java ---
@@ -0,0 +1,153 @@
+/**
+ * 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.kinesis.spout;
+
+import java.io.Serializable;
+
+public class ZkInfo implements Serializable {
+ // comma separated list of zk connect strings to connect to zookeeper e.g. localhost:2181
+ private final String zkUrl;
+ // zk node under which to commit the sequence number of messages. e.g. /committed_sequence_numbers
+ private final String zkNode;
+ // zk session timeout in milliseconds
+ private final Integer sessionTimeoutMs;
+ // zk connection timeout in milliseconds
+ private final Integer connectionTimeoutMs;
+ // interval at which to commit offsets to zk in milliseconds
+ private final Long commitIntervalMs;
+ // number of retry attempts for zk
+ private final Integer retryAttempts;
+ // time to sleep between retries in milliseconds
+ private final Integer retryIntervalMs;
+
+ /**
+ * Default constructor that uses defaults for a local setup
+ */
+ public ZkInfo () {
+ this("localhost:2181", "/kinesisOffsets", 20000, 15000, 10000L, 3, 2000);
--- End diff --
Will remove it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72921486
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
--- End diff --
@priyank5485 looks like this method is going through a switch statement of sort, i.e its checking bunch if conditions and doing actions on that part. It will be great if you can add why you are doing it. For the future reference it will be important as this part is most important on when to ack a message.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72278528
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/Config.java ---
@@ -0,0 +1,166 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class Config implements Serializable {
--- End diff --
given that this is a public class going to be used by devs. Can we rename this as KinesisConfig
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the issue:
https://github.com/apache/storm/pull/1586
+1. Build failures not related to the patch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the issue:
https://github.com/apache/storm/pull/1586
@harshach actually some of the build errors are related see https://issues.apache.org/jira/browse/STORM-2021
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r73055551
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,449 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // object handling zk interaction
+ private transient ZKConnection zkConnection;
+ // object handling interaction with kinesis
+ private transient KinesisConnection kinesisConnection;
+ // Kinesis Spout KinesisConfig object
+ private transient final KinesisConfig kinesisConfig;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+
+ KinesisRecordsManager (KinesisConfig kinesisConfig) {
+ this.kinesisConfig = kinesisConfig;
+ this.zkConnection = new ZKConnection(kinesisConfig.getZkInfo());
+ this.kinesisConnection = new KinesisConnection(kinesisConfig.getKinesisConnectionInfo());
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ kinesisConnection.initialize();
+ zkConnection.initialize();
+ List<Shard> shards = kinesisConnection.getShardsForStream(kinesisConfig.getStreamName());
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = kinesisConfig.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ kinesisConfig.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.warn("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.warn("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < kinesisConfig.getMaxUncommittedRecords())) {
+ LOG.warn("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.warn("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // if an ack is received for a message then add it to the ackedPerShard TreeSet. TreeSet because while committing we need to figure out what is the
+ // highest sequence number that can be committed for this shard
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ // if the acked message was in emittedPerShard that means we need to remove it from the emittedPerShard(which keeps track of in flight tuples)
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ // an acked message should not be in failed since if it fails and gets re-emitted it moves to emittedPerShard from failedPerShard. Defensive coding.
+ // Remove it from failedPerShard anyway
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ // if an ack is for a message that failed once at least and was re-emitted then the record itself will be in failedAndFetchedRecords. We use that to
+ // determine if the FailedMessageRetryHandler needs to be told about it and then remove the record itself to clean up memory
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ kinesisConfig.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (kinesisConfig.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // We have three mutually disjoint treesets per shard at any given time to keep track of what sequence number can be committed to zookeeper.
+ // emittedPerShard, ackedPerShard and failedPerShard. Any record starts by entering emittedPerShard. On ack it moves from emittedPerShard to
+ // ackedPerShard and on fail if retry service tells us to retry then it moves from emittedPerShard to failedPerShard. The failed records will move from
+ // failedPerShard to emittedPerShard when the failed record is emitted again as a retry.
+ // Logic for deciding what sequence number to commit is find the highest sequence number from ackedPerShard called X such that there is no sequence
+ // number Y in emittedPerShard or failedPerShard that satisfies X > Y. For e.g. if ackedPerShard is 1,4,5, emittedPerShard is 2,6 and
+ // failedPerShard is 3,7 then we can only commit 1 and not 4 because 2 is still pending and 3 has failed
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ zkConnection.commitState(kinesisConfig.getStreamName(), shardId, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ kinesisConnection.initialize();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ kinesisConnection.shutdown();
+ }
+
+ void close () {
+ commit();
+ kinesisConnection.shutdown();
+ zkConnection.shutdown();
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = kinesisConnection.fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.warn("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
--- End diff --
I would prefer this to be not configurable. Reason is kinesis api somewhere mentioned in their documentation that if you dont receive any records it is good to sleep for a second. Reason is that they some limits for GetRecords request per second and it will throw an exception if that limit is reached or violated. Although I am catching that exception I wanted to avoid that. Making it configurable and go beyond 1 second can slow down the spout in some cases as well. For example, sometimes it takes multiple requests to get to the portion of the shard where the message at a sequence number you want to read is present. In that case it does not make sense to backoff in an exponential fashion. You just want to make sure you get there sooner without violating the limit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72291643
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
+ return false;
+ }
+ // if first failure add it to the count map
+ if (!failCounts.containsKey(messageId)) {
+ failCounts.put(messageId, 0L);
+ }
+ // increment the fail count as we started with 0
+ Long failCount = failCounts.get(messageId);
+ failCounts.put(messageId, ++failCount);
+ // if fail count is greater than maxRetries, discard or ack. for e.g. for maxRetries 3, 4 failures are allowed at maximum
+ if (failCount > maxRetries) {
+ LOG.debug("maxRetries reached so dropping " + messageId);
--- End diff --
I will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72297573
--- Diff: external/storm-kinesis/README.md ---
@@ -0,0 +1,139 @@
+#Storm Kinesis Spout
+Provides core storm spout for consuming data from a stream in Amazon Kinesis Streams. It stores the sequence numbers that can be committed in zookeeper and
+starts consuming records after that sequence number on restart by default. Below is the code sample to create a sample topology that uses the spout. Each
+object used in configuring the spout is explained below. Ideally, the number of spout tasks should be equal to number of shards in kinesis. However each task
+can read from more than one shard.
+
+```java
+public class KinesisSpoutTopology {
+ public static void main (String args[]) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
+ String topologyName = args[0];
+ RecordToTupleMapper recordToTupleMapper = new TestRecordToTupleMapper();
+ KinesisConnectionInfo kinesisConnectionInfo = new KinesisConnectionInfo(new CredentialsProviderChain(), new ClientConfiguration(), Regions.US_WEST_2,
+ 1000);
+ org.apache.storm.kinesis.spout.Config config = new org.apache.storm.kinesis.spout.Config(args[1], ShardIteratorType.TRIM_HORIZON,
+ recordToTupleMapper, new Date(), new ExponentialBackoffRetrier(), new ZkInfo(), kinesisConnectionInfo, 10000L);
+ KinesisSpout kinesisSpout = new KinesisSpout(config);
+ TopologyBuilder topologyBuilder = new TopologyBuilder();
+ topologyBuilder.setSpout("spout", kinesisSpout, 3);
+ topologyBuilder.setBolt("bolt", new KinesisBoltTest(), 1).shuffleGrouping("spout");
+ Config topologyConfig = new Config();
+ topologyConfig.setDebug(true);
+ topologyConfig.setNumWorkers(3);
+ StormSubmitter.submitTopology(topologyName, topologyConfig, topologyBuilder.createTopology());
+ }
+}
+```
+As you can see above the spout takes an object of Config in its constructor. The constructor of Config takes 8 objects as explained below.
+
+#### `String` streamName
+name of kinesis stream to consume data from
+
+#### `ShardIteratorType` shardIteratorType
+3 types are supported - TRIM_HORIZON(beginning of shard), LATEST and AT_TIMESTAMP. By default this argument is ignored if state for shards
+is found in zookeeper. Hence they will apply the first time a topology is started. If you want to use any of these in subsequent runs of the topology, you
+will need to clear the state of zookeeper node used for storing sequence numbers
+#### `RecordToTupleMapper` recordToTupleMapper
--- End diff --
line break
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72279426
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
+
+ /**
+ * no args constructor that uses defaults of 100 ms for first retry, max retries of Long.MAX_VALUE and an exponential backoff of Math.pow(2,i-1) secs for
+ * retry i where i = 2,3,
+ */
+ public ExponentialBackoffRetrier () {
+ this(100L, 2L, Long.MAX_VALUE);
+ }
+
+ /**
+ *
+ * @param initialDelayMillis delay in milliseconds for first retry
+ * @param baseSeconds base for exponent function in seconds
+ * @param maxRetries maximum number of retries before the record is discarded/acked
+ */
+ public ExponentialBackoffRetrier (Long initialDelayMillis, Long baseSeconds, Long maxRetries) {
+ this.initialDelayMillis = initialDelayMillis;
+ this.baseSeconds = baseSeconds;
+ this.maxRetries = maxRetries;
+ validate();
+ }
+
+ private void validate () {
+ if (initialDelayMillis < 0) {
+ throw new IllegalArgumentException("initialDelayMillis cannot be negative." );
+ }
+ if (baseSeconds < 0) {
+ throw new IllegalArgumentException("baseSeconds cannot be negative.");
+ }
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries cannot be negative.");
+ }
+ }
+ @Override
+ public boolean failed(KinesisMessageId messageId) {
+ LOG.debug("Handling failed message " + messageId);
+ // if maxRetries is 0, dont retry and return false as per interface contract
+ if (maxRetries == 0) {
+ LOG.debug("maxRetries set to 0. Hence not queueing " + messageId);
+ return false;
+ }
+ // if first failure add it to the count map
+ if (!failCounts.containsKey(messageId)) {
+ failCounts.put(messageId, 0L);
+ }
+ // increment the fail count as we started with 0
+ Long failCount = failCounts.get(messageId);
+ failCounts.put(messageId, ++failCount);
+ // if fail count is greater than maxRetries, discard or ack. for e.g. for maxRetries 3, 4 failures are allowed at maximum
+ if (failCount > maxRetries) {
+ LOG.debug("maxRetries reached so dropping " + messageId);
+ failCounts.remove(messageId);
+ return false;
+ }
+ // if reached so far, add it to the set of messages waiting to be retried with next retry time based on how many times it failed
+ retryTimes.put(messageId, getRetryTime(failCount));
+ retryMessageSet.add(messageId);
+ LOG.debug("Scheduled " + messageId + " for retry at " + retryTimes.get(messageId) + " and retry attempt " + failCount);
--- End diff --
I recommend this to be at warn level as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72296325
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
--- End diff --
this should be warn
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72306241
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
--- End diff --
I would prefer this to be not configurable. Reason is kinesis api somewhere mentioned in their documentation that if you dont receive any records it is good to sleep for a second. Reason is that they some limits for GetRecords request per second and it will throw an exception if that limit is reached or violated. Although I am catching that exception I wanted to avoid that. Making it configurable and go beyond 1 second can slow down the spout in some cases as well. For example, sometimes it takes multiple requests to get to the portion of the shard where the message at a sequence number you want to read is present. In that case it does not make sense to backoff in an exponential fashion. You just want to make sure you get there sooner without violating the limit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72296990
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
--- End diff --
if we can document via comments on this ack logic that would be great.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on the issue:
https://github.com/apache/storm/pull/1586
@harshach I added some more comments in ack method and commit method for acking logic. I also moved code interacting with zk and kinesis in to its own class to make KinesisRecordsManager smaller. Please review when you get a chance
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72296861
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
--- End diff --
if we can make multiple returns into one or two we should try.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72292758
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ExponentialBackoffRetrier.java ---
@@ -0,0 +1,164 @@
+/**
+ * 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.kinesis.spout;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+public class ExponentialBackoffRetrier implements FailedMessageRetryHandler, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(ExponentialBackoffRetrier.class);
+ // Wait interfal for retrying after first failure
+ private final Long initialDelayMillis;
+ // Base for exponential function in seconds for retrying for second, third and so on failures
+ private final Long baseSeconds;
+ // Maximum number of retries
+ private final Long maxRetries;
+ // map to track number of failures for each kinesis message that failed
+ private Map<KinesisMessageId, Long> failCounts = new HashMap<>();
+ // map to track next retry time for each kinesis message that failed
+ private Map<KinesisMessageId, Long> retryTimes = new HashMap<>();
+ // sorted set of records to be retrued based on retry time. earliest retryTime record comes first
+ private SortedSet<KinesisMessageId> retryMessageSet = new TreeSet<>(new RetryTimeComparator());
--- End diff --
No. Messages are not stored anywhere in ExponentialBackoffRetrier. retryTimes is just used to implement the comparator interface that will be used by retryMessageSet TreeSet to order the messages on their next retry time. You can check out the RetryTimeComparator in this class
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72300543
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ZkInfo.java ---
@@ -0,0 +1,153 @@
+/**
+ * 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.kinesis.spout;
+
+import java.io.Serializable;
+
+public class ZkInfo implements Serializable {
+ // comma separated list of zk connect strings to connect to zookeeper e.g. localhost:2181
+ private final String zkUrl;
+ // zk node under which to commit the sequence number of messages. e.g. /committed_sequence_numbers
+ private final String zkNode;
+ // zk session timeout in milliseconds
+ private final Integer sessionTimeoutMs;
+ // zk connection timeout in milliseconds
+ private final Integer connectionTimeoutMs;
+ // interval at which to commit offsets to zk in milliseconds
+ private final Long commitIntervalMs;
+ // number of retry attempts for zk
+ private final Integer retryAttempts;
+ // time to sleep between retries in milliseconds
+ private final Integer retryIntervalMs;
+
+ /**
+ * Default constructor that uses defaults for a local setup
+ */
+ public ZkInfo () {
+ this("localhost:2181", "/kinesisOffsets", 20000, 15000, 10000L, 3, 2000);
--- End diff --
I think this is useful for testing but I can see users tripping over this :) so can we remove this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72294493
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisConnectionInfo.java ---
@@ -0,0 +1,137 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.regions.Regions;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Serializable;
+import java.util.Arrays;
+
+public class KinesisConnectionInfo implements Serializable {
+ private final byte[] serializedKinesisCredsProvider;
+ private final byte[] serializedkinesisClientConfig;
+ private final Integer recordsLimit;
+ private final Regions region;
+
+ private transient AWSCredentialsProvider credentialsProvider;
+ private transient ClientConfiguration clientConfiguration;
+
+ /**
+ *
+ * @param credentialsProvider implementation to provide credentials to connect to kinesis
+ * @param clientConfiguration client configuration to pass to kinesis client
+ * @param region region to connect to
+ * @param recordsLimit max records to be fetched in a getRecords request to kinesis
+ */
+ public KinesisConnectionInfo (AWSCredentialsProvider credentialsProvider, ClientConfiguration clientConfiguration, Regions region, Integer recordsLimit) {
+ if (recordsLimit == null || recordsLimit <= 0) {
+ throw new IllegalArgumentException("recordsLimit has to be a positive integer");
+ }
+ if (region == null) {
+ throw new IllegalArgumentException("region cannot be null");
+ }
+ serializedKinesisCredsProvider = getKryoSerializedBytes(credentialsProvider);
+ serializedkinesisClientConfig = getKryoSerializedBytes(clientConfiguration);
+ this.recordsLimit = recordsLimit;
+ this.region = region;
+
+ this.credentialsProvider = null;
--- End diff --
we don't need to explicitly assign a null to these.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72296389
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
--- End diff --
same here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1586: STORM-1839: Storm spout implementation for Amazon Kinesis...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on the issue:
https://github.com/apache/storm/pull/1586
@harshach Thanks for the review. I have addressed your comments and pushed a commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72302997
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisConnectionInfo.java ---
@@ -0,0 +1,137 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.regions.Regions;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.Serializable;
+import java.util.Arrays;
+
+public class KinesisConnectionInfo implements Serializable {
+ private final byte[] serializedKinesisCredsProvider;
+ private final byte[] serializedkinesisClientConfig;
+ private final Integer recordsLimit;
+ private final Regions region;
+
+ private transient AWSCredentialsProvider credentialsProvider;
+ private transient ClientConfiguration clientConfiguration;
+
+ /**
+ *
+ * @param credentialsProvider implementation to provide credentials to connect to kinesis
+ * @param clientConfiguration client configuration to pass to kinesis client
+ * @param region region to connect to
+ * @param recordsLimit max records to be fetched in a getRecords request to kinesis
+ */
+ public KinesisConnectionInfo (AWSCredentialsProvider credentialsProvider, ClientConfiguration clientConfiguration, Regions region, Integer recordsLimit) {
+ if (recordsLimit == null || recordsLimit <= 0) {
+ throw new IllegalArgumentException("recordsLimit has to be a positive integer");
+ }
+ if (region == null) {
+ throw new IllegalArgumentException("region cannot be null");
+ }
+ serializedKinesisCredsProvider = getKryoSerializedBytes(credentialsProvider);
+ serializedkinesisClientConfig = getKryoSerializedBytes(clientConfiguration);
+ this.recordsLimit = recordsLimit;
+ this.region = region;
+
+ this.credentialsProvider = null;
--- End diff --
Will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72306421
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/ZkInfo.java ---
@@ -0,0 +1,153 @@
+/**
+ * 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.kinesis.spout;
+
+import java.io.Serializable;
+
+public class ZkInfo implements Serializable {
+ // comma separated list of zk connect strings to connect to zookeeper e.g. localhost:2181
+ private final String zkUrl;
+ // zk node under which to commit the sequence number of messages. e.g. /committed_sequence_numbers
+ private final String zkNode;
+ // zk session timeout in milliseconds
+ private final Integer sessionTimeoutMs;
+ // zk connection timeout in milliseconds
+ private final Integer connectionTimeoutMs;
+ // interval at which to commit offsets to zk in milliseconds
+ private final Long commitIntervalMs;
+ // number of retry attempts for zk
+ private final Integer retryAttempts;
+ // time to sleep between retries in milliseconds
+ private final Integer retryIntervalMs;
+
+ /**
+ * Default constructor that uses defaults for a local setup
+ */
+ public ZkInfo () {
+ this("localhost:2181", "/kinesisOffsets", 20000, 15000, 10000L, 3, 2000);
+ }
+
+ public ZkInfo (String zkUrl, String zkNode, Integer sessionTimeoutMs, Integer connectionTimeoutMs, Long commitIntervalMs, Integer retryAttempts, Integer
+ retryIntervalMs) {
+ this.zkUrl = zkUrl;
+ this.zkNode = zkNode;
+ this.sessionTimeoutMs = sessionTimeoutMs;
+ this.connectionTimeoutMs = connectionTimeoutMs;
+ this.commitIntervalMs = commitIntervalMs;
+ this.retryAttempts = retryAttempts;
+ this.retryIntervalMs = retryIntervalMs;
+ validate();
+ }
+
+ public String getZkUrl() {
+ return zkUrl;
+ }
+
+ public String getZkNode() {
+ return zkNode;
+ }
+
+ public Integer getSessionTimeoutMs() {
+ return sessionTimeoutMs;
+ }
+
+ public Integer getConnectionTimeoutMs() {
+ return connectionTimeoutMs;
+ }
+
+ public Long getCommitIntervalMs() {
+ return commitIntervalMs;
+ }
+
+ public Integer getRetryAttempts() {
+ return retryAttempts;
+ }
+
+ public Integer getRetryIntervalMs() {
+ return retryIntervalMs;
+ }
+
+ private void validate () {
+
+ if (zkUrl == null || zkUrl.length() < 1) {
+ throw new IllegalArgumentException("zkUrl must be specified to connect to zookeeper");
+ }
+ if (zkNode == null || zkNode.length() < 1) {
+ throw new IllegalArgumentException("zkNode must be specified");
+ }
+ checkPositive(sessionTimeoutMs, "sessionTimeoutMs");
+ checkPositive(connectionTimeoutMs, "connectionTimeoutMs");
+ checkPositive(commitIntervalMs, "commitIntervalMs");
+ checkPositive(retryAttempts, "retryAttempts");
+ checkPositive(retryIntervalMs, "retryIntervalMs");
+ }
+
+ private void checkPositive (Integer argument, String name) {
+ if (argument == null && argument <= 0) {
+ throw new IllegalArgumentException(name + " must be positive");
+ }
+ }
+ private void checkPositive (Long argument, String name) {
+ if (argument == null && argument <= 0) {
+ throw new IllegalArgumentException(name + " must be positive");
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "ZkInfo{" +
+ "zkUrl='" + zkUrl + '\'' +
+ ", zkNode='" + zkNode + '\'' +
+ ", sessionTimeoutMs=" + sessionTimeoutMs +
+ ", connectionTimeoutMs=" + connectionTimeoutMs +
+ ", commitIntervalMs=" + commitIntervalMs +
+ ", retryAttempts=" + retryAttempts +
+ ", retryIntervalMs=" + retryIntervalMs +
+ '}';
+ }
+
+ @Override
--- End diff --
Will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72298893
--- Diff: external/storm-kinesis/src/main/java/org/apache/storm/kinesis/spout/KinesisRecordsManager.java ---
@@ -0,0 +1,566 @@
+/**
+ * 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.kinesis.spout;
+
+import com.amazonaws.regions.Region;
+import com.amazonaws.services.kinesis.AmazonKinesisClient;
+import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
+import com.amazonaws.services.kinesis.model.DescribeStreamResult;
+import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
+import com.amazonaws.services.kinesis.model.GetRecordsRequest;
+import com.amazonaws.services.kinesis.model.GetRecordsResult;
+import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
+import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
+import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
+import com.amazonaws.services.kinesis.model.Record;
+import com.amazonaws.services.kinesis.model.Shard;
+import com.amazonaws.services.kinesis.model.ShardIteratorType;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+class KinesisRecordsManager {
+ private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordsManager.class);
+ // zk interaction object
+ private transient CuratorFramework curatorFramework;
+ // Kinesis Spout Config object
+ private transient final Config config;
+ // Queue of records per shard fetched from kinesis and are waiting to be emitted
+ private transient Map<String, LinkedList<Record>> toEmitPerShard = new HashMap<>();
+ // Map of records that were fetched from kinesis as a result of failure and are waiting to be emitted
+ private transient Map<KinesisMessageId, Record> failedandFetchedRecords = new HashMap<>();
+ // Sequence numbers per shard that have been emitted. LinkedHashSet as we need to remove on ack or fail. At the same time order is needed to figure out the
+ // sequence number to commit. Logic explained in commit
+ private transient Map<String, TreeSet<BigInteger>> emittedPerShard = new HashMap<>();
+ // sorted acked sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> ackedPerShard = new HashMap<>();
+ // sorted failed sequence numbers - needed to figure out what sequence number can be committed
+ private transient Map<String, TreeSet<BigInteger>> failedPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for new messages
+ private transient Map<String, String> shardIteratorPerShard = new HashMap<>();
+ // last fetched sequence number corresponding to position in shard
+ private transient Map<String, String> fetchedSequenceNumberPerShard = new HashMap<>();
+ // shard iterator corresponding to position in shard for failed messages
+ private transient Map<KinesisMessageId, String> shardIteratorPerFailedMessage = new HashMap<>();
+ // timestamp to decide when to commit to zk again
+ private transient long lastCommitTime;
+ // boolean to track deactivated state
+ private transient boolean deactivated;
+ private transient AmazonKinesisClient kinesisClient;
+
+ KinesisRecordsManager (Config config) {
+ this.config = config;
+ }
+
+ void initialize (int myTaskIndex, int totalTasks) {
+ deactivated = false;
+ lastCommitTime = System.currentTimeMillis();
+ initializeKinesisClient();
+ initializeCurator();
+ List<Shard> shards = this.getShards();
+ LOG.info("myTaskIndex is " + myTaskIndex);
+ LOG.info("totalTasks is " + totalTasks);
+ int i = myTaskIndex;
+ while (i < shards.size()) {
+ LOG.info("Shard id " + shards.get(i).getShardId() + " assigned to task " + myTaskIndex);
+ toEmitPerShard.put(shards.get(i).getShardId(), new LinkedList<Record>());
+ i += totalTasks;
+ }
+ initializeFetchedSequenceNumbers();
+ refreshShardIteratorsForNewRecords();
+ }
+
+ void next (SpoutOutputCollector collector) {
+ if (shouldCommit()) {
+ commit();
+ }
+ KinesisMessageId failedMessageId = config.getFailedMessageRetryHandler().getNextFailedMessageToRetry();
+ if (failedMessageId != null) {
+ // if the retry service returns a message that is not in failed set then ignore it. should never happen
+ BigInteger failedSequenceNumber = new BigInteger(failedMessageId.getSequenceNumber());
+ if (failedPerShard.containsKey(failedMessageId.getShardId()) && failedPerShard.get(failedMessageId.getShardId()).contains(failedSequenceNumber)) {
+ if (!failedandFetchedRecords.containsKey(failedMessageId)) {
+ fetchFailedRecords(failedMessageId);
+ }
+ if (emitFailedRecord(collector, failedMessageId)) {
+ failedPerShard.get(failedMessageId.getShardId()).remove(failedSequenceNumber);
+ config.getFailedMessageRetryHandler().failedMessageEmitted(failedMessageId);
+ return;
+ } else {
+ LOG.debug("failedMessageEmitted not called on retrier for " + failedMessageId + ". This can happen a few times but should not happen " +
+ "infinitely");
+ }
+ } else {
+ LOG.debug("failedPerShard does not contain " + failedMessageId + ". This should never happen.");
+ }
+ }
+ LOG.debug("No failed record to emit for now. Hence will try to emit new records");
+ // if maximum uncommitted records count has reached, so dont emit any new records and return
+ if (!(getUncommittedRecordsCount() < config.getMaxUncommittedRecords())) {
+ LOG.debug("maximum uncommitted records count has reached. so not emitting any new records and returning");
+ return;
+ }
+ // early return as no shard is assigned - probably because number of executors > number of shards
+ if (toEmitPerShard.isEmpty()) {
+ LOG.debug("No shard is assigned to this task. Hence not emitting any tuple.");
+ return;
+ }
+
+ if (shouldFetchNewRecords()) {
+ fetchNewRecords();
+ }
+ emitNewRecord(collector);
+ }
+
+ void ack (KinesisMessageId kinesisMessageId) {
+ // for an acked message add it to acked set and remove it from emitted and failed
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Ack received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ if (!ackedPerShard.containsKey(shardId)) {
+ ackedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ ackedPerShard.get(shardId).add(sequenceNumber);
+ if (emittedPerShard.containsKey(shardId)) {
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ }
+ if (failedPerShard.containsKey(shardId)) {
+ failedPerShard.get(shardId).remove(sequenceNumber);
+ }
+ if (failedandFetchedRecords.containsKey(kinesisMessageId)) {
+ config.getFailedMessageRetryHandler().acked(kinesisMessageId);
+ failedandFetchedRecords.remove(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void fail (KinesisMessageId kinesisMessageId) {
+ String shardId = kinesisMessageId.getShardId();
+ BigInteger sequenceNumber = new BigInteger(kinesisMessageId.getSequenceNumber());
+ LOG.debug("Fail received for shardId: " + shardId + " sequenceNumber: " + sequenceNumber);
+ // for a failed message add it to failed set if it will be retried, otherwise ack it; remove from emitted either way
+ if (config.getFailedMessageRetryHandler().failed(kinesisMessageId)) {
+ if (!failedPerShard.containsKey(shardId)) {
+ failedPerShard.put(shardId, new TreeSet<BigInteger>());
+ }
+ failedPerShard.get(shardId).add(sequenceNumber);
+ TreeSet<BigInteger> emitted = emittedPerShard.get(shardId);
+ emitted.remove(sequenceNumber);
+ } else {
+ ack(kinesisMessageId);
+ }
+ // keep committing when topology is deactivated since ack and fail keep getting called on deactivated topology
+ if (deactivated) {
+ commit();
+ }
+ }
+
+ void commit () {
+ // Logic for deciding what sequence number to ack is find the highest sequence number from acked called X such that there is no sequence number Y in
+ // emitted or failed that satisfies X > Y. For e.g. is acked is 1,3,5. Emitted is 2,4,6 then we can only commit 1 and not 3 because 2 is still pending
+ for (String shardId: toEmitPerShard.keySet()) {
+ if (ackedPerShard.containsKey(shardId)) {
+ BigInteger commitSequenceNumberBound = null;
+ if (failedPerShard.containsKey(shardId) && !failedPerShard.get(shardId).isEmpty()) {
+ commitSequenceNumberBound = failedPerShard.get(shardId).first();
+ }
+ if (emittedPerShard.containsKey(shardId) && !emittedPerShard.get(shardId).isEmpty()) {
+ BigInteger smallestEmittedSequenceNumber = emittedPerShard.get(shardId).first();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(smallestEmittedSequenceNumber) == 1)) {
+ commitSequenceNumberBound = smallestEmittedSequenceNumber;
+ }
+ }
+ Iterator<BigInteger> ackedSequenceNumbers = ackedPerShard.get(shardId).iterator();
+ BigInteger ackedSequenceNumberToCommit = null;
+ while (ackedSequenceNumbers.hasNext()) {
+ BigInteger ackedSequenceNumber = ackedSequenceNumbers.next();
+ if (commitSequenceNumberBound == null || (commitSequenceNumberBound.compareTo(ackedSequenceNumber) == 1)) {
+ ackedSequenceNumberToCommit = ackedSequenceNumber;
+ ackedSequenceNumbers.remove();
+ } else {
+ break;
+ }
+ }
+ if (ackedSequenceNumberToCommit != null) {
+ Map<Object, Object> state = new HashMap<>();
+ state.put("committedSequenceNumber", ackedSequenceNumberToCommit.toString());
+ LOG.debug("Committing sequence number " + ackedSequenceNumberToCommit.toString() + " for shardId " + shardId);
+ String path = getZkPath(shardId);
+ commitState(path, state);
+ }
+ }
+ }
+ lastCommitTime = System.currentTimeMillis();
+ }
+
+ void activate () {
+ LOG.info("Activate called");
+ deactivated = false;
+ initializeKinesisClient();
+ }
+
+ void deactivate () {
+ LOG.info("Deactivate called");
+ deactivated = true;
+ commit();
+ shutdownKinesisClient();
+ }
+
+ void close () {
+ commit();
+ shutdownKinesisClient();
+ shutdownCurator();
+ }
+
+ private String getZkPath (String shardId) {
+ String path = "";
+ if (!config.getZkInfo().getZkNode().startsWith("/")) {
+ path += "/";
+ }
+ path += config.getZkInfo().getZkNode();
+ if (!config.getZkInfo().getZkNode().endsWith("/")) {
+ path += "/";
+ }
+ path += (config.getStreamName() + "/" + shardId);
+ return path;
+ }
+
+ private void commitState (String path, Map<Object, Object> state) {
+ byte[] bytes = JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8"));
+ try {
+ if (curatorFramework.checkExists().forPath(path) == null) {
+ curatorFramework.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ curatorFramework.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<Object, Object> readState (String path) {
+ try {
+ Map<Object, Object> state = null;
+ byte[] b = null;
+ if (curatorFramework.checkExists().forPath(path) != null) {
+ b = curatorFramework.getData().forPath(path);
+ }
+ if (b != null) {
+ state = (Map<Object, Object>) JSONValue.parse(new String(b, "UTF-8"));
+ }
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // fetch records from kinesis starting at sequence number for message passed as argument. Any other messages fetched and are in the failed queue will also
+ // be kept in memory to avoid going to kinesis again for retry
+ private void fetchFailedRecords (KinesisMessageId kinesisMessageId) {
+ // if shard iterator not present for this message, get it
+ if (!shardIteratorPerFailedMessage.containsKey(kinesisMessageId)) {
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ }
+ String shardIterator = shardIteratorPerFailedMessage.get(kinesisMessageId);
+ LOG.debug("Fetching failed records for shard id :" + kinesisMessageId.getShardId() + " at sequence number " + kinesisMessageId.getSequenceNumber() +
+ " using shardIterator " + shardIterator);
+ try {
+ GetRecordsResult getRecordsResult = fetchRecords(shardIterator);
+ if (getRecordsResult != null) {
+ List<Record> records = getRecordsResult.getRecords();
+ LOG.debug("Records size from fetchFailedRecords is " + records.size());
+ // update the shard iterator to next one in case this fetch does not give the message.
+ shardIteratorPerFailedMessage.put(kinesisMessageId, getRecordsResult.getNextShardIterator());
+ if (records.size() == 0) {
+ LOG.debug("No records returned from kinesis. Hence sleeping for 1 second");
+ Thread.sleep(1000);
+ } else {
+ // add all fetched records to the set of failed records if they are present in failed set
+ for (Record record: records) {
+ KinesisMessageId current = new KinesisMessageId(kinesisMessageId.getStreamName(), kinesisMessageId.getShardId(), record.getSequenceNumber());
+ if (failedPerShard.get(kinesisMessageId.getShardId()).contains(new BigInteger(current.getSequenceNumber()))) {
+ failedandFetchedRecords.put(current, record);
+ shardIteratorPerFailedMessage.remove(current);
+ }
+ }
+ }
+ }
+ } catch (InterruptedException ie) {
+ LOG.debug("Thread interrupted while sleeping", ie);
+ } catch (ExpiredIteratorException ex) {
+ LOG.debug("shardIterator for failedRecord " + kinesisMessageId + " has expired. Refreshing shardIterator");
+ refreshShardIteratorForFailedRecord(kinesisMessageId);
+ } catch (ProvisionedThroughputExceededException pe) {
+ try {
+ LOG.debug("ProvisionedThroughputExceededException occured. Check your limits. Sleeping for 1 second.", pe);
+ Thread.sleep(1000);
--- End diff --
same as before
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1586: STORM-1839: Storm spout implementation for Amazon ...
Posted by priyank5485 <gi...@git.apache.org>.
Github user priyank5485 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1586#discussion_r72305457
--- Diff: external/storm-kinesis/README.md ---
@@ -0,0 +1,139 @@
+#Storm Kinesis Spout
+Provides core storm spout for consuming data from a stream in Amazon Kinesis Streams. It stores the sequence numbers that can be committed in zookeeper and
+starts consuming records after that sequence number on restart by default. Below is the code sample to create a sample topology that uses the spout. Each
+object used in configuring the spout is explained below. Ideally, the number of spout tasks should be equal to number of shards in kinesis. However each task
+can read from more than one shard.
+
+```java
+public class KinesisSpoutTopology {
+ public static void main (String args[]) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
+ String topologyName = args[0];
+ RecordToTupleMapper recordToTupleMapper = new TestRecordToTupleMapper();
+ KinesisConnectionInfo kinesisConnectionInfo = new KinesisConnectionInfo(new CredentialsProviderChain(), new ClientConfiguration(), Regions.US_WEST_2,
+ 1000);
+ org.apache.storm.kinesis.spout.Config config = new org.apache.storm.kinesis.spout.Config(args[1], ShardIteratorType.TRIM_HORIZON,
+ recordToTupleMapper, new Date(), new ExponentialBackoffRetrier(), new ZkInfo(), kinesisConnectionInfo, 10000L);
+ KinesisSpout kinesisSpout = new KinesisSpout(config);
+ TopologyBuilder topologyBuilder = new TopologyBuilder();
+ topologyBuilder.setSpout("spout", kinesisSpout, 3);
+ topologyBuilder.setBolt("bolt", new KinesisBoltTest(), 1).shuffleGrouping("spout");
+ Config topologyConfig = new Config();
+ topologyConfig.setDebug(true);
+ topologyConfig.setNumWorkers(3);
+ StormSubmitter.submitTopology(topologyName, topologyConfig, topologyBuilder.createTopology());
+ }
+}
+```
+As you can see above the spout takes an object of Config in its constructor. The constructor of Config takes 8 objects as explained below.
+
+#### `String` streamName
+name of kinesis stream to consume data from
+
+#### `ShardIteratorType` shardIteratorType
+3 types are supported - TRIM_HORIZON(beginning of shard), LATEST and AT_TIMESTAMP. By default this argument is ignored if state for shards
+is found in zookeeper. Hence they will apply the first time a topology is started. If you want to use any of these in subsequent runs of the topology, you
+will need to clear the state of zookeeper node used for storing sequence numbers
+#### `RecordToTupleMapper` recordToTupleMapper
--- End diff --
Will change it
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---