Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-167: Fix consumer start for close/open shards

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-aws/issues/167

A combo of closing / opening shards leads to consumer not starting
This commit is contained in:
Anton Malinovskiy
2020-08-10 20:16:59 +02:00
committed by GitHub
parent 10e96e2de2
commit e0df55a6c3
2 changed files with 76 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -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();

86
src/test/java/org/springframework/integration/aws/inbound/KinesisMessageDrivenChannelAdapterTests.java
View File

@@ -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);