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

GH-190: Swallow checkpoint provisioning exception

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-aws/issues/190

Swallows `ProvisionedThroughputExceededException` while checkpointing
exhausted shards to avoid the `ShardConsumer` from not being marked as
closed and therefore be left in an inconsistent state which will only
throw exceptions as the `shardIterator` would be `null` and the
`ShardConsumer` wouldn't be marked as `CLOSED`.
This commit is contained in:
acm19
2021-01-12 16:01:46 +01:00
committed by GitHub
parent 696472037f
commit b43a230ffe
2 changed files with 29 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2020 the original author or authors.
* Copyright 2017-2021 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.
@@ -93,6 +93,7 @@ import com.amazonaws.services.kinesis.model.ShardIteratorType;
* @author Hervé Fortin
* @author Dirk Bonhomme
* @author Greg Eales
* @author Asiel Caballero
*
* @since 1.1
*/
@@ -1091,7 +1092,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
String endingSequenceNumber =
shard.getSequenceNumberRange().getEndingSequenceNumber();
if (endingSequenceNumber != null) {
this.checkpointer.checkpoint(endingSequenceNumber);
checkpointSwallowingProvisioningExceptions(endingSequenceNumber);
}
break;
}
@@ -1125,6 +1126,19 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport
};
}
private void checkpointSwallowingProvisioningExceptions(String endingSequenceNumber) {
try {
this.checkpointer.checkpoint(endingSequenceNumber);
}
catch (ProvisionedThroughputExceededException ignored) {
// This exception is ignored to gurantee that an exhausted shard is marked as CLOSED
// even in the case it's not possible to checkpoint. Otherwise the ShardConsumer is
// left in an illegal state where the shard iterator is null without any possibility
// of recovering from it.
logger.debug("Exception while checkpointing empty shards", ignored);
}
}
private GetRecordsResult getRecords(GetRecordsRequest getRecordsRequest) {
try {
return KinesisMessageDrivenChannelAdapter.this.amazonKinesis.getRecords(getRecordsRequest);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2020 the original author or authors.
* Copyright 2017-2021 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.
@@ -75,6 +75,7 @@ import com.amazonaws.services.kinesis.model.Shard;
* @author Artem Bilan
* @author Matthias Wesolowski
* @author Greg Eales
* @author Asiel Caballero
*
* @since 1.1
*/
@@ -237,7 +238,6 @@ public class KinesisMessageDrivenChannelAdapterTests {
@Configuration
@EnableIntegration
public static class Config {
private final AtomicReference<KinesisShardEndedEvent> shardEndedEventReference = new AtomicReference<>();
@Bean
@@ -440,7 +440,7 @@ public class KinesisMessageDrivenChannelAdapterTests {
@Bean
public ConcurrentMetadataStore reshardingCheckpointStore() {
return new SimpleMetadataStore();
return new ExceptionReadyMetadataStore();
}
@Bean
@@ -472,4 +472,14 @@ public class KinesisMessageDrivenChannelAdapterTests {
}
private static class ExceptionReadyMetadataStore extends SimpleMetadataStore {
@Override
public boolean replace(String key, String oldValue, String newValue) {
if ("SpringIntegration:streamForResharding:closedShard4".equals(key)) {
throw new ProvisionedThroughputExceededException("Throughput exceeded");
}
return super.replace(key, oldValue, newValue);
}
}
}