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

No NextShardIterator if not processed manual ack batch

Browse Source 
* fix(KinesisMessageDrivenChannelAdapter): added logic to not use the NextShardIterator if Manual Checkpointer hasnt reached the last checkpoint

* fix(KinesisMessageDrivenChannel): added logs for getNextShardIterator

* fix(KinesisMessageDrivenChannelAdapter): fixed logic and comments issue

* feat(KinesisMessageDrivenChannelAdapter): implemented integration tests for getNextShardIterator new usecase

* feat(KinesisMessageDrivenChannelAdapter): updated @author

* feat(KinesisMessageDrivenChannelAdapterTests): added author data
This commit is contained in:
Jonathan Nagayoshi
2021-05-25 14:52:53 -03:00
committed by GitHub
parent f10d6c6620
commit a002646f63
2 changed files with 100 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -94,6 +94,7 @@ import com.amazonaws.services.kinesis.model.ShardIteratorType;
* @author Dirk Bonhomme
* @author Greg Eales
* @author Asiel Caballero
* @author Jonathan Nagayoshi
*
* @since 1.1
*/
@@ -1073,7 +1074,40 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
finally {
attributesHolder.remove();
if (result != null) {
this.shardIterator = result.getNextShardIterator();
// If using manual checkpointer, we have to make sure we are allowed to use the next shard iterator
// Because if the manual checkpointer was not set to the latest record, it means there are records to be reprocessed
// and if we use the nextShardIterator, we will be skipping records that need to be reprocessed
List<Record> records = result.getRecords();
if (CheckpointMode.manual.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode) &&
!records.isEmpty()) {
logger.info("Manual checkpointer. Must validate if should use getNextShardIterator()");
String lastRecordSequence = records.get(records.size() - 1).getSequenceNumber();
String lastCheckpointSequence = this.checkpointer.getCheckpoint();
if (lastCheckpointSequence.equals(lastRecordSequence)) {
logger.info("latestCheckpointSequence is same as latestRecordSequence. " +
"" +
"Should getNextShardIterator()");
// Means the manual checkpointer has processed the last record, Should move forward
this.shardIterator = result.getNextShardIterator();
}
else {
logger.info("latestCheckpointSequence is not the same as latestRecordSequence" +
". Should Get a new iterator AFTER_SEQUENCE_NUMBER latestCheckpointSequence");
// Something wrong happened and not all records were processed.
// Must start from the latest known checkpoint
KinesisShardOffset newOffset = new KinesisShardOffset(this.shardOffset);
newOffset.setSequenceNumber(lastCheckpointSequence);
newOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER);
GetShardIteratorRequest shardIteratorRequest = newOffset.toShardIteratorRequest();
this.shardIterator = KinesisMessageDrivenChannelAdapter.this
.amazonKinesis
.getShardIterator(shardIteratorRequest)
.getShardIterator();
}
}
else {
this.shardIterator = result.getNextShardIterator();
}
if (this.shardIterator == null) {
if (KinesisMessageDrivenChannelAdapter.this.lockRegistry != null) {

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

@@ -76,6 +76,7 @@ import com.amazonaws.services.kinesis.model.Shard;
* @author Matthias Wesolowski
* @author Greg Eales
* @author Asiel Caballero
* @author Jonathan Nagayoshi
*
* @since 1.1
*/
@@ -201,6 +202,40 @@ public class KinesisMessageDrivenChannelAdapterTests {
.hasSize(2);
this.kinesisMessageDrivenChannelAdapter.stop();
this.kinesisMessageDrivenChannelAdapter.setListenerMode(ListenerMode.batch);
this.kinesisMessageDrivenChannelAdapter.setCheckpointMode(CheckpointMode.manual);
this.checkpointStore.put("SpringIntegration" + ":" + STREAM1 + ":" + "1", "2");
this.kinesisMessageDrivenChannelAdapter.start();
message = this.kinesisChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isInstanceOf(List.class);
List<String> messagePayload = (List<String>) message.getPayload();
assertThat(messagePayload).size().isEqualTo(3);
Object messageSequenceNumberHeader = message.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER);
assertThat(messageSequenceNumberHeader).isInstanceOf(List.class);
assertThat((List<String>) messageSequenceNumberHeader).contains("3");
// Set checkpoint to 3, this should prevent adapter from using next shard, since its not the latest record
// in the batch
checkpointer.checkpoint("3");
await().untilAsserted(
() -> assertThat(this.checkpointStore.get("SpringIntegration" + ":" + STREAM1 + ":" + "1"))
.isEqualTo("3"));
message = this.kinesisChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isInstanceOf(List.class);
messagePayload = (List<String>) message.getPayload();
assertThat(messagePayload).size().isEqualTo(2);
assertThat(messagePayload).contains("bar");
assertThat(messagePayload).contains("foobar");
this.kinesisMessageDrivenChannelAdapter.stop();
}
@Test
@@ -296,6 +331,36 @@ public class KinesisMessageDrivenChannelAdapterTests {
.withRecords(new Record().withPartitionKey("partition1").withSequenceNumber("2")
.withData(ByteBuffer.wrap(serializingConverter.convert("bar")))));
String shard1Iterator5 = "shard1Iterator5";
String shard1Iterator6 = "shard1Iterator6";
given(amazonKinesis.getShardIterator(
KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "2").toShardIteratorRequest()))
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator5));
given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator5).withLimit(25)))
.willReturn(new GetRecordsResult().withNextShardIterator(shard1Iterator6)
.withRecords(new Record().withPartitionKey("partition1").withSequenceNumber("3")
.withData(ByteBuffer.wrap(serializingConverter.convert("foo"))),
new Record().withPartitionKey("partition1").withSequenceNumber("4")
.withData(ByteBuffer.wrap(serializingConverter.convert("bar"))),
new Record().withPartitionKey("partition1").withSequenceNumber("5")
.withData(ByteBuffer.wrap(serializingConverter.convert("foobar")))));
given(amazonKinesis.getShardIterator(
KinesisShardOffset.afterSequenceNumber(STREAM1, "1", "3").toShardIteratorRequest()))
.willReturn(new GetShardIteratorResult().withShardIterator(shard1Iterator6));
given(amazonKinesis.getRecords(new GetRecordsRequest().withShardIterator(shard1Iterator6).withLimit(25)))
.willReturn(new GetRecordsResult().withNextShardIterator(shard1Iterator6)
.withRecords(
new Record().withPartitionKey("partition1").withSequenceNumber("4")
.withData(ByteBuffer.wrap(serializingConverter.convert("bar"))),
new Record().withPartitionKey("partition1").withSequenceNumber("5")
.withData(ByteBuffer.wrap(serializingConverter.convert("foobar")))));
return amazonKinesis;
}