This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24156 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 7fdc747b086addfef4c5b454b1b31116a47f8774 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 17:20:25 2026 +0200 CAMEL-24156: camel-aws2-kinesis - Fix shardClosed NPE and expired iterator stall - Initialize shardClosed field to its documented default (ignore) so handleClosedShard does not NPE on first shard rotation with default config - Handle ExpiredIteratorException by invalidating the cached iterator and requesting a fresh one, preventing permanent consumer stall after >5min gaps Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../aws2/kinesis/Kinesis2Configuration.java | 2 +- .../component/aws2/kinesis/Kinesis2Consumer.java | 58 ++++++++++++++++------ 2 files changed, 43 insertions(+), 17 deletions(-) diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java index 4b214f845fec..fe6942707717 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java +++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Configuration.java @@ -68,7 +68,7 @@ public class Kinesis2Configuration implements Cloneable, AwsCommonConfiguration + " In case of ignore a WARN message will be logged once and the consumer will not process new messages until restarted," + "in case of silent there will be no logging and the consumer will not process new messages until restarted," + "in case of fail a ReachedClosedStateException will be thrown") - private Kinesis2ShardClosedStrategyEnum shardClosed; + private Kinesis2ShardClosedStrategyEnum shardClosed = Kinesis2ShardClosedStrategyEnum.ignore; @UriParam(label = "proxy", enums = "HTTP,HTTPS", defaultValue = "HTTPS", description = "To define a proxy protocol when instantiating the Kinesis client") private Protocol proxyProtocol = Protocol.HTTPS; diff --git a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java index c08263fd6b13..e452792ccbac 100644 --- a/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java +++ b/components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Consumer.java @@ -44,6 +44,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest; import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse; +import software.amazon.awssdk.services.kinesis.model.ExpiredIteratorException; import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest; import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse; import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest; @@ -173,6 +174,42 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R return; } + GetRecordsResponse result; + try { + result = getRecords(shardIterator, kinesisConnection); + } catch (ExpiredIteratorException e) { + LOG.warn("Shard iterator expired for shard {} on stream {}, requesting a fresh one", + shard.shardId(), getEndpoint().getConfiguration().getStreamName()); + currentShardIterators.remove(shard.shardId()); + try { + shardIterator = getShardIterator(shard, kinesisConnection); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new RuntimeException(ie); + } catch (ExecutionException ee) { + throw new RuntimeException(ee); + } + if (ObjectHelper.isEmpty(shardIterator)) { + return; + } + result = getRecords(shardIterator, kinesisConnection); + } + + try { + Queue<Exchange> exchanges = createExchanges(shard, result.records()); + processedExchangeCount.addAndGet(processBatch(CastUtils.cast(exchanges))); + } catch (Exception e) { + throw new RuntimeException(e); + } + + // May cache the last successful sequence number, and pass it to the + // getRecords request. That way, on the next poll, we start from where + // we left off, however, I don't know what happens to subsequent + // exchanges when an earlier exchange fails. + updateShardIterator(shard, result.nextShardIterator()); + } + + private GetRecordsResponse getRecords(String shardIterator, KinesisConnection kinesisConnection) { GetRecordsRequest req = GetRecordsRequest .builder() .shardIterator(shardIterator) @@ -181,10 +218,9 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R .getMaxResultsPerRequest()) .build(); - GetRecordsResponse result; if (getEndpoint().getConfiguration().isAsyncClient()) { try { - result = kinesisConnection + return kinesisConnection .getAsyncClient(getEndpoint()) .getRecords(req) .get(); @@ -192,26 +228,16 @@ public class Kinesis2Consumer extends ScheduledBatchPollingConsumer implements R Thread.currentThread().interrupt(); throw new RuntimeException(e); } catch (ExecutionException e) { + if (e.getCause() instanceof ExpiredIteratorException ex) { + throw ex; + } throw new RuntimeException(e); } } else { - result = kinesisConnection + return kinesisConnection .getClient(getEndpoint()) .getRecords(req); } - - try { - Queue<Exchange> exchanges = createExchanges(shard, result.records()); - processedExchangeCount.addAndGet(processBatch(CastUtils.cast(exchanges))); - } catch (Exception e) { - throw new RuntimeException(e); - } - - // May cache the last successful sequence number, and pass it to the - // getRecords request. That way, on the next poll, we start from where - // we left off, however, I don't know what happens to subsequent - // exchanges when an earlier exchange fails. - updateShardIterator(shard, result.nextShardIterator()); } private void updateShardIterator(Shard shard, String nextShardIterator) {
