flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tzulitai <...@git.apache.org>
Subject [GitHub] flink pull request #2131: [FLINK-3231][streaming-connectors] FlinkKinesisCon...
Date Fri, 24 Jun 2016 14:49:34 GMT
Github user tzulitai commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2131#discussion_r68409331
  
    --- Diff: flink-streaming-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java
---
    @@ -159,42 +217,65 @@ public String getShardIterator(KinesisStreamShard shard, String
shardIteratorTyp
     		return kinesisClient.getShardIterator(shard.getStreamName(), shard.getShardId(), shardIteratorType,
startingSeqNum).getShardIterator();
     	}
     
    +	private List<KinesisStreamShard> getShardsOfStream(String streamName, String lastSeenShardId)
throws InterruptedException {
    +		List<KinesisStreamShard> shardsOfStream = new ArrayList<>();
    +
    +		DescribeStreamResult describeStreamResult;
    +		do {
    +			describeStreamResult = describeStream(streamName, lastSeenShardId);
    +
    +			List<Shard> shards = describeStreamResult.getStreamDescription().getShards();
    +			for (Shard shard : shards) {
    +				shardsOfStream.add(new KinesisStreamShard(streamName, shard));
    +			}
    +
    +			if (shards.size() != 0) {
    +				lastSeenShardId = shards.get(shards.size() - 1).getShardId();
    +			}
    +		} while (describeStreamResult.getStreamDescription().isHasMoreShards());
    +
    +		return shardsOfStream;
    +	}
    +
     	/**
     	 * Get metainfo for a Kinesis stream, which contains information about which shards
this Kinesis stream possess.
     	 *
    +	 * This method is using a "full jitter" approach described in
    +	 * <a href="http://google.com">https://www.awsarchitectureblog.com/2015/03/backoff.html</a>.
This is necessary
    +	 * because concurrent calls will be made by all parallel subtask's {@link ShardDiscoverer}s.
This jitter backoff
    +	 * approach will help distribute calls across the discoverers over time.
    +	 *
     	 * @param streamName the stream to describe
     	 * @param startShardId which shard to start with for this describe operation (earlier
shard's infos will not appear in result)
     	 * @return the result of the describe stream operation
     	 */
    -	private DescribeStreamResult describeStream(String streamName, String startShardId)
{
    +	private DescribeStreamResult describeStream(String streamName, String startShardId)
throws InterruptedException {
     		final DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
     		describeStreamRequest.setStreamName(streamName);
     		describeStreamRequest.setExclusiveStartShardId(startShardId);
     
     		DescribeStreamResult describeStreamResult = null;
     		String streamStatus = null;
    -		int remainingRetryTimes = Integer.valueOf(
    -			configProps.getProperty(KinesisConfigConstants.CONFIG_STREAM_DESCRIBE_RETRIES, Integer.toString(KinesisConfigConstants.DEFAULT_STREAM_DESCRIBE_RETRY_TIMES)));
    -		long describeStreamBackoffTimeInMillis = Long.valueOf(
    -			configProps.getProperty(KinesisConfigConstants.CONFIG_STREAM_DESCRIBE_BACKOFF, Long.toString(KinesisConfigConstants.DEFAULT_STREAM_DESCRIBE_BACKOFF)));
     
    -		// Call DescribeStream, with backoff and retries (if we get LimitExceededException).
    -		while ((remainingRetryTimes >= 0) && (describeStreamResult == null)) {
    +		// Call DescribeStream, with full-jitter backoff (if we get LimitExceededException).
    +		Random seed = null;
    +		int attemptCount = 0;
    +		while (describeStreamResult == null) { // retry until we get a result
     			try {
     				describeStreamResult = kinesisClient.describeStream(describeStreamRequest);
     				streamStatus = describeStreamResult.getStreamDescription().getStreamStatus();
     			} catch (LimitExceededException le) {
    -				LOG.warn("Got LimitExceededException when describing stream " + streamName + ". Backing
off for "
    -					+ describeStreamBackoffTimeInMillis + " millis.");
    -				try {
    -					Thread.sleep(describeStreamBackoffTimeInMillis);
    -				} catch (InterruptedException ie) {
    -					LOG.debug("Stream " + streamName + " : Sleep  was interrupted ", ie);
    +				if (seed == null) {
    +					seed = new Random();
     				}
    +				long backoffMillis = fullJitterBackoff(
    +					describeStreamBaseBackoffMillis, describeStreamMaxBackoffMillis, describeStreamExpConstant,
attemptCount++, seed);
    +				LOG.warn("Got LimitExceededException when describing stream " + streamName + ". Backing
off for "
    +					+ backoffMillis + " millis.");
    +				Thread.sleep(backoffMillis);
     			} catch (ResourceNotFoundException re) {
     				throw new RuntimeException("Error while getting stream details", re);
     			}
    -			remainingRetryTimes--;
     		}
     
     		if (streamStatus == null) {
    --- End diff --
    
    Actually, since we'll be retrying until we get a describeStreamResult now, this message
and RuntimeException can be removed.


---
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.
---

Mime
View raw message