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

GH-63: Kinesis: Add diagnostics for closed shards

Browse Source 
Fixes spring-projects/spring-integration-aws#63
This commit is contained in:
Artem Bilan
2017-04-06 13:37:35 -04:00
parent 8e7e1a02b6
commit d65fbebc31
3 changed files with 145 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Shard> 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<Record> 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:

2
src/main/java/org/springframework/integration/aws/inbound/kinesis/ShardCheckpointer.java
View File

@@ -79,7 +79,7 @@ class ShardCheckpointer implements Checkpointer {
List<Record> filterRecords(List<Record> records) {
List<Record> recordsToProcess = new LinkedList<>(records);
this.lastCheckpointValue = this.checkpointStore.get(this.key);
this.lastCheckpointValue = getCheckpoint();
if (this.lastCheckpointValue != null) {
for (Iterator<Record> iterator = recordsToProcess.iterator(); iterator.hasNext(); ) {
Record record = iterator.next();

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

@@ -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<KinesisShardOffset> 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;
}
}