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 3062f5b..b5325e6 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 @@ -1594,7 +1594,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } private boolean renewLockInRegistry(LockCompletableFuture renewLockFuture) { - if (KinesisMessageDrivenChannelAdapter.this.lockRegistry instanceof RenewableLockRegistry renewableLockRegistry) { + if (KinesisMessageDrivenChannelAdapter.this.lockRegistry + instanceof RenewableLockRegistry renewableLockRegistry) { + try { renewableLockRegistry.renewLock(renewLockFuture.lockKey); return renewLockFuture.complete(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 35a3e93..bb814d8 100644 --- a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java +++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2018-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. diff --git a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java index 4674e59..89bf492 100644 --- a/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java +++ b/src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRepository.java @@ -112,7 +112,7 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, /** * Default value for the {@link #leaseDuration} property. */ - public static final Duration DEFAULT_LEASE_DURATION = Duration.ofSeconds(10); + public static final Duration DEFAULT_LEASE_DURATION = Duration.ofSeconds(60); private static final Log LOGGER = LogFactory.getLog(DynamoDbLockRegistry.class); @@ -309,7 +309,7 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, .withProjectionExpression(KEY_ATTR) .withMaxResultSize(1) .withFilterExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " >= :ttl") - .withValueMap(ownerWithCurrentTimeValues()); + .withValueMap(ownerWithTtlValues(currentEpochSeconds())); return this.lockTable.query(querySpec).iterator().hasNext(); } @@ -355,15 +355,15 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, new DeleteItemSpec() .withPrimaryKey(KEY_ATTR, lock) .withConditionExpression(OWNER_ATTR + " = :owner AND " + TTL_ATTR + " < :ttl") - .withValueMap(ownerWithCurrentTimeValues()))); + .withValueMap(ownerWithTtlValues(currentEpochSeconds())))); this.heldLocks.clear(); } } - private ValueMap ownerWithCurrentTimeValues() { + private ValueMap ownerWithTtlValues(long epochSeconds) { ValueMap valueMap = new ValueMap() - .withNumber(":ttl", currentEpochSeconds()); + .withNumber(":ttl", epochSeconds); valueMap.putAll(this.ownerAttribute); return valueMap; } @@ -375,16 +375,17 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, */ public boolean acquire(String lock) { awaitForActive(); + long currentTime = currentEpochSeconds(); PutItemSpec putItemSpec = new PutItemSpec() .withItem( new Item() .withPrimaryKey(KEY_ATTR, lock) .withString(OWNER_ATTR, this.owner) - .withLong(CREATED_ATTR, currentEpochSeconds()) + .withLong(CREATED_ATTR, currentTime) .withLong(TTL_ATTR, ttlEpochSeconds())) .withConditionExpression(LOCK_NOT_EXISTS_EXPRESSION) - .withValueMap(ownerWithCurrentTimeValues()); + .withValueMap(ownerWithTtlValues(currentTime)); try { this.lockTable.putItem(putItemSpec); this.heldLocks.add(lock); @@ -393,7 +394,6 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, catch (ConditionalCheckFailedException ex) { return false; } - } /** @@ -404,16 +404,12 @@ public class DynamoDbLockRepository implements InitializingBean, DisposableBean, public boolean renew(String lock) { awaitForActive(); if (this.heldLocks.contains(lock)) { - ValueMap valueMap = - new ValueMap() - .withNumber(":ttl", ttlEpochSeconds()); - valueMap.putAll(this.ownerAttribute); UpdateItemSpec updateItemSpec = new UpdateItemSpec() .withPrimaryKey(KEY_ATTR, lock) .withUpdateExpression("SET " + TTL_ATTR + " = :ttl") .withConditionExpression(LOCK_EXISTS_EXPRESSION) - .withValueMap(valueMap); + .withValueMap(ownerWithTtlValues(ttlEpochSeconds())); try { this.lockTable.updateItem(updateItemSpec); return true; diff --git a/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java b/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java index 788f746..4ae4bde 100644 --- a/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java +++ b/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java @@ -39,6 +39,7 @@ import org.testcontainers.utility.DockerImageName; * started only once per JVM, therefore the target Docker container is reused automatically. * * @author Artem Bilan + * * @since 3.0 */ @Testcontainers(disabledWithoutDocker = true) @@ -46,7 +47,7 @@ public interface LocalstackContainerTest { LocalStackContainer LOCAL_STACK_CONTAINER = new LocalStackContainer( - DockerImageName.parse("localstack/localstack:1.2.0")) + DockerImageName.parse("localstack/localstack:1.4.0")) .withServices( LocalStackContainer.Service.DYNAMODB, LocalStackContainer.Service.KINESIS, diff --git a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java b/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java index 2788b21..088af81 100644 --- a/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java +++ b/src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java @@ -17,6 +17,9 @@ package org.springframework.integration.aws.lock; import java.time.Duration; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; @@ -25,12 +28,14 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; +import com.amazonaws.services.dynamodbv2.model.AttributeValue; import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest; import com.amazonaws.waiters.FixedDelayStrategy; import com.amazonaws.waiters.MaxAttemptsRetryStrategy; import com.amazonaws.waiters.PollingStrategy; import com.amazonaws.waiters.Waiter; import com.amazonaws.waiters.WaiterParameters; +import org.assertj.core.data.Percentage; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -341,10 +346,20 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest { assertThat(lock.tryLock()).isTrue(); try { + this.dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(60)); assertThatNoException().isThrownBy(() -> this.dynamoDbLockRegistry.renewLock("foo")); + String ttl = + DYNAMO_DB.getItem(DynamoDbLockRepository.DEFAULT_TABLE_NAME, + Map.of(DynamoDbLockRepository.KEY_ATTR, new AttributeValue("foo"))) + .getItem() + .get(DynamoDbLockRepository.TTL_ATTR).getN(); + assertThat(Long.parseLong(ttl)) + .isCloseTo(LocalDateTime.now().plusSeconds(60).toEpochSecond(ZoneOffset.UTC), + Percentage.withPercentage(10)); } finally { lock.unlock(); + this.dynamoDbLockRepository.setLeaseDuration(Duration.ofSeconds(2)); } }