GH-219: Fix DynamoDbLockReg for skip blocking

Fixes https://github.com/spring-projects/spring-integration-aws/issues/219

Turns out the `AmazonDynamoDBLockClient` doesn't have a proper logic to
determine correct `lookupTime` and if we want to skip blocking waits,
the item is always treated as not expired because just obtained item from DB
is updated to the current time for its `lookupTime` property

* Remove the logic in the `DynamoDbLockRegistry` setting `withShouldSkipBlockingWait(true)`

**Cherry-pick to `2.5.x`**

src/main/java/org/springframework/integration/aws/lock/DynamoDbLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 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.
@@ -450,8 +450,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
 		private void setupDefaultAcquireLockOptionsBuilder() {
 			this.acquireLockOptionsBuilder
 					.withAdditionalTimeToWaitForLock(Long.MAX_VALUE - DynamoDbLockRegistry.this.leaseDuration)
-					.withRefreshPeriod(DynamoDbLockRegistry.this.refreshPeriod)
-					.withShouldSkipBlockingWait(false);
+					.withRefreshPeriod(DynamoDbLockRegistry.this.refreshPeriod);
 		}
 
 		@Override
@@ -505,13 +504,7 @@ public class DynamoDbLockRegistry implements ExpirableLockRegistry, Initializing
 			long additionalTimeToWait = Math
 					.max(TimeUnit.MILLISECONDS.convert(time, unit) - System.currentTimeMillis() + start, 0L);
 
-			this.acquireLockOptionsBuilder.withAdditionalTimeToWaitForLock(additionalTimeToWait)
-					.withRefreshPeriod(DynamoDbLockRegistry.this.refreshPeriod)
-					.withShouldSkipBlockingWait(false);
-
-			if (additionalTimeToWait == 0) {
-				this.acquireLockOptionsBuilder.withShouldSkipBlockingWait(true);
-			}
+			this.acquireLockOptionsBuilder.withAdditionalTimeToWaitForLock(additionalTimeToWait);
 
 			boolean acquired = false;
 			try {

src/test/java/org/springframework/integration/aws/lock/DynamoDbLockRegistryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2022 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.aws.lock;
 
+import java.lang.reflect.Method;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
@@ -23,7 +24,11 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 
+import com.amazonaws.services.dynamodbv2.AcquireLockOptions;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClient;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBLockClientOptions;
+import com.amazonaws.services.dynamodbv2.LockItem;
 import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
 import com.amazonaws.waiters.FixedDelayStrategy;
 import com.amazonaws.waiters.MaxAttemptsRetryStrategy;
@@ -43,6 +48,7 @@ import org.springframework.integration.aws.LocalstackContainerTest;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -316,6 +322,42 @@ public class DynamoDbLockRegistryTests implements LocalstackContainerTest {
 		assertThat(((Exception) imse).getMessage()).contains("You do not own");
 	}
 
+	@Test
+	void abandonedLock() throws Exception {
+		Method awaitForActiveMethod = ReflectionUtils.findMethod(DynamoDbLockRegistry.class, "awaitForActive");
+		ReflectionUtils.makeAccessible(awaitForActiveMethod);
+		ReflectionUtils.invokeMethod(awaitForActiveMethod, this.dynamoDbLockRegistry);
+
+		AmazonDynamoDBLockClientOptions lockClientOptions =
+				AmazonDynamoDBLockClientOptions.builder(DYNAMO_DB, DynamoDbLockRegistry.DEFAULT_TABLE_NAME)
+						.withPartitionKeyName(DynamoDbLockRegistry.DEFAULT_PARTITION_KEY_NAME)
+						.withSortKeyName(DynamoDbLockRegistry.DEFAULT_SORT_KEY_NAME)
+						.withCreateHeartbeatBackgroundThread(false)
+						.withLeaseDuration(2L)
+						.build();
+		AmazonDynamoDBLockClient lockClient = new AmazonDynamoDBLockClient(lockClientOptions);
+
+		AcquireLockOptions lockOptions =
+				AcquireLockOptions.builder("foo")
+				.withReplaceData(false)
+				.withSortKey(DynamoDbLockRegistry.DEFAULT_SORT_KEY)
+				.build();
+
+		LockItem lockItem = lockClient.acquireLock(lockOptions);
+		assertThat(lockItem).isNotNull();
+
+		lockClient.close();
+
+		Lock lock = this.dynamoDbLockRegistry.obtain("foo");
+		int n = 0;
+		while (!lock.tryLock() && n++ < 100) {
+			Thread.sleep(100);
+		}
+
+		assertThat(n).isLessThan(100);
+		lock.unlock();
+	}
+
 	@Configuration
 	public static class ContextConfiguration {