From b552a0892ccc23f33a2c6caeff5893524dd9dd59 Mon Sep 17 00:00:00 2001 From: Krzysztof Witkowski Date: Sun, 26 Mar 2017 21:48:12 +0200 Subject: [PATCH] Handle expired and throttling in ShardConsumer When shard iterator is expired (`ExpiredIteratorException`), we have to restart from the stored sequence number When shard iterator is throttled (`ProvisionedThroughputExceededException`), we have to "sleep" for backoff period * Catch `ExpiredIteratorException` on the `.amazonKinesis.getRecords()` request and move `ShardConsumer` to the `ConsumerState.EXPIRED`. This state is treated as `NEW`, therefore a fresh `amazonKinesis.getShardIterator()` is performed * Catch `ProvisionedThroughputExceededException` on the `.amazonKinesis.getRecords()` request and move `ShardConsumer` to the `ConsumerState.SLEEP` for the `KinesisMessageDrivenChannelAdapter.this.consumerBackoff` period --- .../KinesisMessageDrivenChannelAdapter.java | 86 +++++++++++++------ ...nesisMessageDrivenChannelAdapterTests.java | 36 +++++--- 2 files changed, 83 insertions(+), 39 deletions(-) diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index a55056e..90ccaa2 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -56,10 +56,12 @@ import org.springframework.util.StringUtils; import com.amazonaws.services.kinesis.AmazonKinesis; 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.LimitExceededException; +import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException; import com.amazonaws.services.kinesis.model.Record; import com.amazonaws.services.kinesis.model.ResourceNotFoundException; import com.amazonaws.services.kinesis.model.Shard; @@ -675,11 +677,12 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i switch (this.state) { case NEW: + case EXPIRED: this.task = new Runnable() { @Override public void run() { - if (logger.isInfoEnabled()) { + if (logger.isInfoEnabled() && ShardConsumer.this.state == ConsumerState.NEW) { logger.info("The [" + this + "] has been started."); } if (ShardConsumer.this.shardOffset.isReset()) { @@ -750,34 +753,34 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); getRecordsRequest.setShardIterator(ShardConsumer.this.shardIterator); getRecordsRequest.setLimit(KinesisMessageDrivenChannelAdapter.this.recordsLimit); - GetRecordsResult result = KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .getRecords(getRecordsRequest); - List records = result.getRecords(); + GetRecordsResult result = getRecords(getRecordsRequest); - if (!records.isEmpty()) { - processRecords(records); - } + if (result != null) { + List records = result.getRecords(); - ShardConsumer.this.shardIterator = result.getNextShardIterator(); - - if (ShardConsumer.this.shardIterator == null) { - // Shard is closed: nothing to consume any more. - // Resharding is possible. - ShardConsumer.this.state = ConsumerState.STOP; - } - - if (ConsumerState.STOP != ShardConsumer.this.state && records.isEmpty()) { - if (logger.isDebugEnabled()) { - logger.debug("No records for [" + ShardConsumer.this + - "] on sequenceNumber [" + - ShardConsumer.this.checkpointer.getLastCheckpointValue() + - "]. Suspend consuming for [" + - KinesisMessageDrivenChannelAdapter.this.consumerBackoff + "] milliseconds."); + if (!records.isEmpty()) { + processRecords(records); + } + + ShardConsumer.this.shardIterator = result.getNextShardIterator(); + + if (ShardConsumer.this.shardIterator == null) { + // Shard is closed: nothing to consume any more. + // Resharding is possible. + ShardConsumer.this.state = ConsumerState.STOP; + } + + if (ConsumerState.STOP != ShardConsumer.this.state && records.isEmpty()) { + if (logger.isDebugEnabled()) { + logger.debug("No records for [" + ShardConsumer.this + + "] on sequenceNumber [" + + ShardConsumer.this.checkpointer.getLastCheckpointValue() + + "]. Suspend consuming for [" + + KinesisMessageDrivenChannelAdapter.this.consumerBackoff + "] milliseconds."); + } + prepareSleepState(); } - ShardConsumer.this.sleepUntil = System.currentTimeMillis() + - KinesisMessageDrivenChannelAdapter.this.consumerBackoff; - ShardConsumer.this.state = ConsumerState.SLEEP; } ShardConsumer.this.task = null; @@ -786,6 +789,37 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i }; } + private GetRecordsResult getRecords(GetRecordsRequest getRecordsRequest) { + try { + return KinesisMessageDrivenChannelAdapter.this.amazonKinesis.getRecords(getRecordsRequest); + } + catch (ExpiredIteratorException e) { + // Iterator expired, but this does not mean that shard no longer contains records. + // Lets acquire iterator again (using checkpointer for iterator start sequence number). + if (logger.isInfoEnabled()) { + logger.info("Shard iterator " + getRecordsRequest.getShardIterator() + " expired.\n" + + "A new one will be started from the check pointed sequence number."); + } + this.state = ConsumerState.EXPIRED; + } + catch (ProvisionedThroughputExceededException e) { + if (logger.isWarnEnabled()) { + logger.warn("GetRecords request throttled for iterator " + getRecordsRequest.getShardIterator() + + " with the reason: " + e.getErrorMessage()); + } + // We are throttled, so let's sleep + prepareSleepState(); + } + + return null; + } + + private void prepareSleepState() { + ShardConsumer.this.sleepUntil = System.currentTimeMillis() + + KinesisMessageDrivenChannelAdapter.this.consumerBackoff; + ShardConsumer.this.state = ConsumerState.SLEEP; + } + private void processRecords(List records) { records = this.checkpointer.filterRecords(records); if (!records.isEmpty()) { @@ -848,7 +882,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private enum ConsumerState { - NEW, CONSUME, SLEEP, STOP + NEW, EXPIRED, CONSUME, SLEEP, STOP } diff --git a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java index c6e7801..2e0338e 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java @@ -54,9 +54,11 @@ import org.springframework.test.context.junit4.SpringRunner; import com.amazonaws.services.kinesis.AmazonKinesis; 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.GetShardIteratorResult; +import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException; import com.amazonaws.services.kinesis.model.Record; import com.amazonaws.services.kinesis.model.SequenceNumberRange; import com.amazonaws.services.kinesis.model.Shard; @@ -195,27 +197,35 @@ public class KinesisMessageDrivenChannelAdapterTests { .withSequenceNumberRange(new SequenceNumberRange() .withEndingSequenceNumber("1"))))); - String shardIterator1 = "shardIterator1"; + String shard1Iterator1 = "shard1Iterator1"; + String shard1Iterator2 = "shard1Iterator2"; given(amazonKinesis.getShardIterator(KinesisShardOffset.latest(STREAM1, "1").toShardIteratorRequest())) - .willReturn(new GetShardIteratorResult() - .withShardIterator(shardIterator1)); + .willReturn( + new GetShardIteratorResult().withShardIterator(shard1Iterator1), + new GetShardIteratorResult().withShardIterator(shard1Iterator2)); - String shardIterator2 = "shardIterator2"; + String shard2Iterator1 = "shard2Iterator1"; given(amazonKinesis.getShardIterator(KinesisShardOffset.latest(STREAM1, "2").toShardIteratorRequest())) .willReturn(new GetShardIteratorResult() - .withShardIterator(shardIterator2)); + .withShardIterator(shard2Iterator1)); + + given(amazonKinesis.getRecords(new GetRecordsRequest() + .withShardIterator(shard1Iterator1) + .withLimit(25))) + .willThrow(new ProvisionedThroughputExceededException("Iterator throttled")) + .willThrow(new ExpiredIteratorException("Iterator expired")); SerializingConverter serializingConverter = new SerializingConverter(); - String shardIterator11 = "shardIterator11"; + String shard1Iterator3 = "shard1Iterator3"; given(amazonKinesis.getRecords(new GetRecordsRequest() - .withShardIterator(shardIterator1) + .withShardIterator(shard1Iterator2) .withLimit(25))) .willReturn(new GetRecordsResult() - .withNextShardIterator(shardIterator11) + .withNextShardIterator(shard1Iterator3) .withRecords(new Record() .withPartitionKey("partition1") .withSequenceNumber("1") @@ -226,21 +236,21 @@ public class KinesisMessageDrivenChannelAdapterTests { .withData(ByteBuffer.wrap(serializingConverter.convert("bar"))))); given(amazonKinesis.getRecords(new GetRecordsRequest() - .withShardIterator(shardIterator2) + .withShardIterator(shard2Iterator1) .withLimit(25))) .willReturn(new GetRecordsResult() - .withNextShardIterator(shardIterator2)); + .withNextShardIterator(shard2Iterator1)); given(amazonKinesis.getRecords(new GetRecordsRequest() - .withShardIterator(shardIterator11) + .withShardIterator(shard1Iterator3) .withLimit(25))) .willReturn(new GetRecordsResult() - .withNextShardIterator(shardIterator11)); + .withNextShardIterator(shard1Iterator3)); given(amazonKinesis.getShardIterator(KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "1") .toShardIteratorRequest())) .willReturn(new GetShardIteratorResult() - .withShardIterator(shardIterator1)); + .withShardIterator(shard1Iterator2)); return amazonKinesis; }