diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index ff613c9..530cdbd 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -161,6 +161,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport private int describeStreamRetries = 50; + private long lockRenewalTimeout = 10_000L; + private boolean resetCheckpoints; private InboundMessageMapper embeddedHeadersMapper; @@ -291,6 +293,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.startTimeout = startTimeout; } + /** + * Configure a timeout in milliseconds to wait for lock on shard renewal. + * @param lockRenewalTimeout the timeout to wait for lock renew in milliseconds. + * @since 2.3.5 + */ + public void setLockRenewalTimeout(long lockRenewalTimeout) { + Assert.isTrue(lockRenewalTimeout > 0, "'lockRenewalTimeout' must be more than 0"); + this.lockRenewalTimeout = lockRenewalTimeout; + } + /** * The maximum number of concurrent {@link ConsumerInvoker}s running. The {@link ShardConsumer}s * are evenly distributed between {@link ConsumerInvoker}s. Messages from within the same shard @@ -920,7 +932,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport LockCompletableFuture unlockFuture = new LockCompletableFuture(this.key); KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.unlock(unlockFuture); try { - unlockFuture.get(1, TimeUnit.SECONDS); + unlockFuture.get(KinesisMessageDrivenChannelAdapter.this.lockRenewalTimeout, TimeUnit.MILLISECONDS); } catch (Exception ex) { if (ex instanceof InterruptedException) { @@ -1029,7 +1041,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport KinesisMessageDrivenChannelAdapter.this.shardConsumerManager.renewLock(renewLockFuture); boolean lockRenewed = false; try { - lockRenewed = renewLockFuture.get(1, TimeUnit.SECONDS); + lockRenewed = renewLockFuture.get(KinesisMessageDrivenChannelAdapter.this.lockRenewalTimeout, + TimeUnit.MILLISECONDS); } catch (Exception ex) { if (ex instanceof InterruptedException) {