From e0df55a6c3e1138b13491afd41c591f7583203f9 Mon Sep 17 00:00:00 2001 From: Anton Malinovskiy Date: Mon, 10 Aug 2020 20:16:59 +0200 Subject: [PATCH] GH-167: Fix consumer start for close/open shards Fixes https://github.com/spring-projects/spring-integration-aws/issues/167 A combo of closing / opening shards leads to consumer not starting --- .../KinesisMessageDrivenChannelAdapter.java | 6 +- ...nesisMessageDrivenChannelAdapterTests.java | 86 ++++++++++++++++--- 2 files changed, 76 insertions(+), 16 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 e17cc65..5929414 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 @@ -708,14 +708,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.consumerExecutor.execute(consumerInvoker); } else { + boolean consumerAdded = false; for (ConsumerInvoker consumerInvoker : this.consumerInvokers) { if (consumerInvoker.consumers.size() < this.consumerInvokerMaxCapacity) { consumerInvoker.addConsumer(shardConsumer); - return; + consumerAdded = true; + break; } } - if (this.concurrency != 0) { + if (this.concurrency != 0 && !consumerAdded) { ConsumerInvoker firstConsumerInvoker = this.consumerInvokers.get(0); firstConsumerInvoker.addConsumer(shardConsumer); this.consumerInvokerMaxCapacity = firstConsumerInvoker.consumers.size(); 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 c7da48a..7b4a1b3 100644 --- a/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java +++ b/src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -207,11 +207,12 @@ public class KinesisMessageDrivenChannelAdapterTests { Map shardConsumers = TestUtils.getPropertyValue(this.reshardingChannelAdapter, "shardConsumers", Map.class); int n = 0; - while (!shardConsumers.isEmpty() && n++ < 100) { + while (shardConsumers.size() != 4 && n++ < 100) { Thread.sleep(100); } assertThat(n).isLessThan(100); + // When resharding happens the describeStream() is performed again verify(this.amazonKinesisForResharding, atLeast(1)) .listShards(any(ListShardsRequest.class)); @@ -222,7 +223,7 @@ public class KinesisMessageDrivenChannelAdapterTests { assertThat(kinesisShardEndedEvent).isNotNull() .extracting(KinesisShardEndedEvent::getShardKey) - .isEqualTo("SpringIntegration:streamForResharding:closedShard"); + .isEqualTo("SpringIntegration:streamForResharding:closedShard4"); } @Configuration @@ -329,25 +330,81 @@ public class KinesisMessageDrivenChannelAdapterTests { public AmazonKinesis amazonKinesisForResharding() { AmazonKinesis amazonKinesis = mock(AmazonKinesis.class); + // kinesis handles adding a shard by closing a shard and opening 2 new instead, creating a scenario where it + // happens couple of times given(amazonKinesis.listShards(new ListShardsRequest().withStreamName(STREAM_FOR_RESHARDING))) .willReturn(new ListShardsResult() - .withShards(new Shard().withShardId("closedShard").withSequenceNumberRange( - new SequenceNumberRange().withEndingSequenceNumber("1")))); + .withShards( + new Shard().withShardId("closedShard1") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("1")))) + .willReturn(new ListShardsResult() + .withShards( + new Shard().withShardId("closedShard1") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("1")), + new Shard().withShardId("newShard2") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("2")), + new Shard().withShardId("newShard3") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("3")), + new Shard().withShardId("closedShard4") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("4")))) + .willReturn(new ListShardsResult() + .withShards( + new Shard().withShardId("closedShard1") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("1")), + new Shard().withShardId("newShard2") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("2")), + new Shard().withShardId("newShard3") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("3")), + new Shard().withShardId("closedShard4") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("4")), + new Shard().withShardId("newShard5") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("5")), + new Shard().withShardId("newShard6") + .withSequenceNumberRange(new SequenceNumberRange().withEndingSequenceNumber("6")))); - 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())))); + setClosedShard(amazonKinesis, "1"); + setNewShard(amazonKinesis, "2"); + setNewShard(amazonKinesis, "3"); + setClosedShard(amazonKinesis, "4"); + setNewShard(amazonKinesis, "5"); + setNewShard(amazonKinesis, "6"); return amazonKinesis; } + private void setClosedShard(AmazonKinesis amazonKinesis, String shardIndex) { + String shardIterator = String.format("shard%sIterator1", shardIndex); + + given(amazonKinesis.getShardIterator( + KinesisShardOffset.latest(STREAM_FOR_RESHARDING, "closedShard" + shardIndex).toShardIteratorRequest())) + .willReturn(new GetShardIteratorResult().withShardIterator(shardIterator)); + + given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shardIterator).withLimit(25))) + .willReturn(new GetRecordsResult().withNextShardIterator(null) + .withRecords(new Record().withPartitionKey("partition1").withSequenceNumber(shardIndex) + .withData(ByteBuffer.wrap("foo".getBytes())))); + } + + private void setNewShard(AmazonKinesis amazonKinesis, String shardIndex) { + String shardIterator1 = String.format("shard%sIterator1", shardIndex); + String shardIterator2 = String.format("shard%sIterator2", shardIndex); + + given(amazonKinesis.getShardIterator( + KinesisShardOffset.latest(STREAM_FOR_RESHARDING, "newShard" + shardIndex).toShardIteratorRequest())) + .willReturn(new GetShardIteratorResult().withShardIterator(shardIterator1)); + + given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shardIterator2).withLimit(25))) + .willReturn(new GetRecordsResult().withNextShardIterator(shardIterator2) + .withRecords(new Record().withPartitionKey("partition1").withSequenceNumber(shardIndex) + .withData(ByteBuffer.wrap("foo".getBytes())))); + + + given(amazonKinesis.getShardIterator( + KinesisShardOffset.latest(STREAM_FOR_RESHARDING, "newShard" + shardIndex).toShardIteratorRequest())) + .willReturn(new GetShardIteratorResult().withShardIterator(shardIterator2)); + } + @Bean public KinesisMessageDrivenChannelAdapter reshardingChannelAdapter() { KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter( @@ -357,6 +414,7 @@ public class KinesisMessageDrivenChannelAdapterTests { adapter.setStartTimeout(10000); adapter.setDescribeStreamRetries(1); adapter.setRecordsLimit(25); + adapter.setConcurrency(1); DirectFieldAccessor dfa = new DirectFieldAccessor(adapter); dfa.setPropertyValue("describeStreamBackoff", 10);