From d65fbebc313ffeb41a4b7b3bbeaa4365e5263bfd Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Apr 2017 13:37:35 -0400 Subject: [PATCH] GH-63: Kinesis: Add diagnostics for closed shards Fixes spring-projects/spring-integration-aws#63 --- .../KinesisMessageDrivenChannelAdapter.java | 59 +++++++--- .../inbound/kinesis/ShardCheckpointer.java | 2 +- ...nesisMessageDrivenChannelAdapterTests.java | 110 ++++++++++++++++-- 3 files changed, 145 insertions(+), 26 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 04537e1..5766468 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 @@ -16,6 +16,7 @@ package org.springframework.integration.aws.inbound.kinesis; +import java.math.BigInteger; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -317,7 +318,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private void restartShardConsumerForOffset(KinesisShardOffset shardOffset) { Assert.isTrue(this.shardOffsets.contains(shardOffset), - "The [" + this + + "The [" + KinesisMessageDrivenChannelAdapter.this + "] doesn't operate shard [" + shardOffset.getShard() + "] for stream [" + shardOffset.getStream() + "]"); @@ -410,12 +411,13 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i try { if (!shardsGatherLatch.await(this.startTimeout, TimeUnit.MILLISECONDS)) { throw new IllegalStateException("The [ " - + this + + + KinesisMessageDrivenChannelAdapter.this + "] could not start during timeout: " + this.startTimeout); } } catch (InterruptedException e) { - throw new IllegalStateException("The [ " + this + "] has been interrupted from start."); + throw new IllegalStateException("The [ " + KinesisMessageDrivenChannelAdapter.this + + "] has been interrupted from start."); } } @@ -470,10 +472,29 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i List shards = describeStreamResult.getStreamDescription().getShards(); for (Shard shard : shards) { - // Check if the shard is still open. Open shards do not have an ending sequence number. - if (shard.getSequenceNumberRange().getEndingSequenceNumber() == null) { - shardsToConsume.add(shard); + String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); + if (endingSequenceNumber != null) { + String key = buildCheckpointKeyForShard(stream, shard.getShardId()); + String checkpoint = KinesisMessageDrivenChannelAdapter.this.checkpointStore.get(key); + + boolean skipClosedShard = checkpoint != null && + new BigInteger(endingSequenceNumber) + .compareTo(new BigInteger(checkpoint)) <= 0; + + if (logger.isTraceEnabled()) { + logger.trace("The shard [" + shard + "] in stream [" + stream + + "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber + + "].\nThe last processed checkpoint is [" + checkpoint + "]." + + (skipClosedShard ? "\nThe shard will be skipped." : "")); + } + + if (skipClosedShard) { + // Skip CLOSED shard which has been read before according a checkpoint + continue; + } } + + shardsToConsume.add(shard); } if (describeStreamResult.getStreamDescription().getHasMoreShards()) { @@ -522,7 +543,6 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i this.resetCheckpoints = false; } - private void populateConsumer(KinesisShardOffset shardOffset) { ShardConsumer shardConsumer = new ShardConsumer(shardOffset); this.shardConsumers.put(shardOffset, shardConsumer); @@ -552,6 +572,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } } + private String buildCheckpointKeyForShard(String stream, String shardId) { + return KinesisMessageDrivenChannelAdapter.this.consumerGroup + ":" + stream + ":" + shardId; + } + @Override protected void doStop() { this.active = false; @@ -645,9 +669,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i ShardConsumer(KinesisShardOffset shardOffset) { this.shardOffset = new KinesisShardOffset(shardOffset); - String key = KinesisMessageDrivenChannelAdapter.this.consumerGroup + - ":" + shardOffset.getStream() + - ":" + shardOffset.getShard(); + String key = buildCheckpointKeyForShard(shardOffset.getStream(), shardOffset.getShard()); this.checkpointer = new ShardCheckpointer(KinesisMessageDrivenChannelAdapter.this.checkpointStore, key); } @@ -671,7 +693,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i @Override public void run() { if (logger.isInfoEnabled() && ShardConsumer.this.state == ConsumerState.NEW) { - logger.info("The [" + this + "] has been started."); + logger.info("The [" + ShardConsumer.this + "] has been started."); } if (ShardConsumer.this.shardOffset.isReset()) { ShardConsumer.this.checkpointer.remove(); @@ -714,13 +736,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i case STOP: if (this.shardIterator == null) { if (logger.isInfoEnabled()) { - logger.info("Stopping the [" + this + + logger.info("Stopping the [" + ShardConsumer.this + + "] on the checkpoint [" + this.checkpointer.getCheckpoint() + "] because the shard has been CLOSED and exhausted."); } } else { if (logger.isInfoEnabled()) { - logger.info("Stopping the [" + this + "]."); + logger.info("Stopping the [" + ShardConsumer.this + "]."); } } this.task = null; @@ -785,15 +808,15 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i // 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" + + logger.info("Shard iterator for [" + ShardConsumer.this + "] 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()); + logger.warn("GetRecords request throttled for [" + ShardConsumer.this + + "] with the reason: " + e.getErrorMessage()); } // We are throttled, so let's sleep prepareSleepState(); @@ -811,8 +834,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private void processRecords(List records) { records = this.checkpointer.filterRecords(records); if (!records.isEmpty()) { - if (logger.isDebugEnabled()) { - logger.debug("Processing records: " + records + " for [" + this + "]"); + if (logger.isTraceEnabled()) { + logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]"); } switch (KinesisMessageDrivenChannelAdapter.this.listenerMode) { case record: diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java index 17ced33..ff5e2f4 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java @@ -79,7 +79,7 @@ class ShardCheckpointer implements Checkpointer { List filterRecords(List records) { List recordsToProcess = new LinkedList<>(records); - this.lastCheckpointValue = this.checkpointStore.get(this.key); + this.lastCheckpointValue = getCheckpoint(); if (this.lastCheckpointValue != null) { for (Iterator iterator = recordsToProcess.iterator(); iterator.hasNext(); ) { Record record = iterator.next(); 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 2e0338e..3fc242d 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java @@ -18,13 +18,17 @@ package org.springframework.integration.aws.inbound; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import java.nio.ByteBuffer; import java.util.List; import java.util.Map; import java.util.Set; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -75,8 +79,10 @@ public class KinesisMessageDrivenChannelAdapterTests { private static final String STREAM1 = "stream1"; + private static final String STREAM_FOR_RESHARDING = "streamForResharding"; + @Autowired - private PollableChannel kinesisChannel; + private QueueChannel kinesisChannel; @Autowired private KinesisMessageDrivenChannelAdapter kinesisMessageDrivenChannelAdapter; @@ -84,9 +90,21 @@ public class KinesisMessageDrivenChannelAdapterTests { @Autowired private MetadataStore checkpointStore; + @Autowired + private KinesisMessageDrivenChannelAdapter reshardingChannelAdapter; + + @Autowired + private AmazonKinesis amazonKinesisForResharding; + + @Before + public void setup() { + this.kinesisChannel.purge(null); + } + @Test @SuppressWarnings({ "unchecked", "rawtypes" }) public void testKinesisMessageDrivenChannelAdapter() throws InterruptedException { + this.kinesisMessageDrivenChannelAdapter.start(); final Set shardOffsets = TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter, "shardOffsets", Set.class); @@ -165,9 +183,30 @@ public class KinesisMessageDrivenChannelAdapterTests { List consumerInvoker = TestUtils.getPropertyValue(this.kinesisMessageDrivenChannelAdapter, "consumerInvokers", List.class); assertThat(consumerInvoker.size()).isEqualTo(2); + + this.kinesisMessageDrivenChannelAdapter.stop(); } + @Test + @SuppressWarnings("rawtypes") + public void testReshadring() throws InterruptedException { + this.reshardingChannelAdapter.start(); + + assertThat(this.kinesisChannel.receive(10000)).isNotNull(); + + Map shardConsumers = TestUtils.getPropertyValue(this.reshardingChannelAdapter, "shardConsumers", Map.class); + + int n = 0; + while (!shardConsumers.isEmpty() && n++ < 100) { + Thread.sleep(100); + } + assertThat(n).isLessThan(100); + + // When resharding happens the describeStream() is performed again + verify(this.amazonKinesisForResharding, times(2)).describeStream(any(DescribeStreamRequest.class)); + } + @Configuration @EnableIntegration public static class Config { @@ -202,8 +241,8 @@ public class KinesisMessageDrivenChannelAdapterTests { given(amazonKinesis.getShardIterator(KinesisShardOffset.latest(STREAM1, "1").toShardIteratorRequest())) .willReturn( - new GetShardIteratorResult().withShardIterator(shard1Iterator1), - new GetShardIteratorResult().withShardIterator(shard1Iterator2)); + new GetShardIteratorResult().withShardIterator(shard1Iterator1), + new GetShardIteratorResult().withShardIterator(shard1Iterator2)); String shard2Iterator1 = "shard2Iterator1"; @@ -212,10 +251,10 @@ public class KinesisMessageDrivenChannelAdapterTests { .withShardIterator(shard2Iterator1)); given(amazonKinesis.getRecords(new GetRecordsRequest() - .withShardIterator(shard1Iterator1) - .withLimit(25))) - .willThrow(new ProvisionedThroughputExceededException("Iterator throttled")) - .willThrow(new ExpiredIteratorException("Iterator expired")); + .withShardIterator(shard1Iterator1) + .withLimit(25))) + .willThrow(new ProvisionedThroughputExceededException("Iterator throttled")) + .willThrow(new ExpiredIteratorException("Iterator expired")); SerializingConverter serializingConverter = new SerializingConverter(); @@ -267,6 +306,7 @@ public class KinesisMessageDrivenChannelAdapterTests { public KinesisMessageDrivenChannelAdapter kinesisMessageDrivenChannelAdapter() { KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter(amazonKinesis(), STREAM1); + adapter.setAutoStartup(false); adapter.setOutputChannel(kinesisChannel()); adapter.setCheckpointStore(checkpointStore()); adapter.setCheckpointMode(CheckpointMode.manual); @@ -276,6 +316,7 @@ public class KinesisMessageDrivenChannelAdapterTests { DirectFieldAccessor dfa = new DirectFieldAccessor(adapter); dfa.setPropertyValue("describeStreamBackoff", 10); + dfa.setPropertyValue("consumerBackoff", 10); return adapter; } @@ -285,6 +326,61 @@ public class KinesisMessageDrivenChannelAdapterTests { return new QueueChannel(); } + @Bean + public AmazonKinesis amazonKinesisForResharding() { + AmazonKinesis amazonKinesis = mock(AmazonKinesis.class); + + given(amazonKinesis.describeStream(new DescribeStreamRequest().withStreamName(STREAM_FOR_RESHARDING))) + .willReturn( + new DescribeStreamResult() + .withStreamDescription(new StreamDescription() + .withStreamName(STREAM_FOR_RESHARDING) + .withStreamStatus(StreamStatus.ACTIVE) + .withHasMoreShards(false) + .withShards(new Shard() + .withShardId("closedShard") + .withSequenceNumberRange(new SequenceNumberRange() + .withEndingSequenceNumber("1"))))); + + String shard1Iterator1 = "shard1Iterator1"; + + given(amazonKinesis.getShardIterator(KinesisShardOffset.latest(STREAM_FOR_RESHARDING, "closedShard") + .toShardIteratorRequest())) + .willReturn( + new GetShardIteratorResult() + .withShardIterator(shard1Iterator1)); + + + given(amazonKinesis.getRecords(new GetRecordsRequest() + .withShardIterator(shard1Iterator1) + .withLimit(25))) + .willReturn(new GetRecordsResult() + .withNextShardIterator(null) + .withRecords(new Record() + .withPartitionKey("partition1") + .withSequenceNumber("1") + .withData(ByteBuffer.wrap("foo".getBytes())))); + + return amazonKinesis; + } + + @Bean + public KinesisMessageDrivenChannelAdapter reshardingChannelAdapter() { + KinesisMessageDrivenChannelAdapter adapter = + new KinesisMessageDrivenChannelAdapter(amazonKinesisForResharding(), STREAM_FOR_RESHARDING); + adapter.setAutoStartup(false); + adapter.setOutputChannel(kinesisChannel()); + adapter.setStartTimeout(10000); + adapter.setDescribeStreamRetries(1); + + DirectFieldAccessor dfa = new DirectFieldAccessor(adapter); + dfa.setPropertyValue("describeStreamBackoff", 10); + dfa.setPropertyValue("consumerBackoff", 10); + + adapter.setConverter(String::new); + + return adapter; + } }