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

Handle expired and throttling in ShardConsumer

Browse Source 
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
This commit is contained in:
Krzysztof Witkowski
2017-03-26 21:48:12 +02:00
committed by Artem Bilan
parent 8d120e020c
commit b552a0892c
2 changed files with 83 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<Record> records = result.getRecords();
GetRecordsResult result = getRecords(getRecordsRequest);
if (!records.isEmpty()) {
processRecords(records);
}
if (result != null) {
List<Record> 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<Record> 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
}

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

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