From e4a72cee17ed8eaeb82baad33ab0599d3a7c27d7 Mon Sep 17 00:00:00 2001 From: abilan Date: Tue, 10 Jan 2023 10:50:55 -0500 Subject: [PATCH] Some tweaks for DynamoDB locks heartbeat Related to https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/180 * Treat non-positive `DynamoDbLockRegistry.heartbeatPeriod` as no heartbeat. This way locks renewal is a responsibility of the target `DynamoDbLockRegistry` consumer. For example, the `KinesisMessageDrivenChannelAdapter` does call `tryLock()` in a loop for locks on shards it is consuming at the moment * Increase locks loop sleep timeout in the `KinesisMessageDrivenChannelAdapter` to one second to avoid many requests to DynamoDB **Cherry-pick to `2.5.x`** # Conflicts: # src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java --- .../KinesisMessageDrivenChannelAdapter.java | 4 ++-- .../aws/lock/DynamoDbLockRegistry.java | 16 ++++++++-------- 2 files changed, 10 insertions(+), 10 deletions(-) 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 4d0a895..e1f65cb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2023 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. @@ -1561,7 +1561,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } - sleep(250, + sleep(1000, new IllegalStateException("ShardConsumerManager Thread [" + this + "] has been interrupted"), true); } diff --git a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java index 91f2339..b9b5c7c 100644 --- a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java +++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java @@ -26,7 +26,6 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; @@ -138,12 +137,6 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing private long heartbeatPeriod = 5L; - /** - * Flag to denote whether the {@link ExecutorService} was provided via the setter and - * thus should not be shutdown when {@link #destroy()} is called. - */ - private boolean executorExplicitlySet; - private volatile boolean initialized; public DynamoDbLockRegistry(AmazonDynamoDB dynamoDB) { @@ -205,6 +198,11 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing this.leaseDuration = leaseDuration; } + /** + * Specify a period in milliseconds how often send locks renewal requests called heartbeat. + * When the value is less than or equal to {@code 0}, the heartbeat is disabled. + * @param heartbeatPeriod the heartbeat period for background thread to renew locks in DB + */ public void setHeartbeatPeriod(long heartbeatPeriod) { this.heartbeatPeriod = heartbeatPeriod; } @@ -228,7 +226,9 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing if (!this.dynamoDBLockClientExplicitlySet) { AmazonDynamoDBLockClientOptions dynamoDBLockClientOptions = AmazonDynamoDBLockClientOptions .builder(this.dynamoDB, this.tableName).withPartitionKeyName(this.partitionKey) - .withSortKeyName(this.sortKeyName).withHeartbeatPeriod(this.heartbeatPeriod) + .withSortKeyName(this.sortKeyName) + .withCreateHeartbeatBackgroundThread(this.heartbeatPeriod > 0) + .withHeartbeatPeriod(this.heartbeatPeriod) .withLeaseDuration(this.leaseDuration).build(); this.dynamoDBLockClient = new AmazonDynamoDBLockClient(dynamoDBLockClientOptions);