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

Some tweaks for DynamoDB locks heartbeat

Browse Source 
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`**
This commit is contained in:
abilan
2023-01-10 10:50:55 -05:00
parent 5d9b464f1b
commit 07603e4aa1
2 changed files with 10 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 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);
}

16
src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java
View File

@@ -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;
@@ -137,12 +136,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) {
@@ -204,6 +197,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;
}
@@ -227,7 +225,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);