GH-3700: Fix Redis lock retry until expiration

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

Fixed a case where lock acquisition attempt was abandoned earlier than expected time.
If a vm uses `tryLock()` and another vm unlock it, it will wake up with an unlock message and try to acquire the lock, 
but if it fails(the other vm obtain lock), vm will return false.

spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 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.
@@ -359,31 +359,39 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 		}
 
 		private boolean subscribeLock(long time) throws ExecutionException, InterruptedException {
-			if (!obtainLock()) {
-				if (!RedisLockRegistry.this.redisMessageListenerContainer.isRunning()) {
-					RedisLockRegistry.this.redisMessageListenerContainer.afterPropertiesSet();
-					RedisLockRegistry.this.redisMessageListenerContainer.start();
-				}
+			final long expiredTime = System.currentTimeMillis() + time;
+			if (obtainLock()) {
+				return true;
+			}
+
+			if (!RedisLockRegistry.this.redisMessageListenerContainer.isRunning()) {
+				RedisLockRegistry.this.redisMessageListenerContainer.afterPropertiesSet();
+				RedisLockRegistry.this.redisMessageListenerContainer.start();
+			}
+			while (time == -1 || expiredTime >= System.currentTimeMillis()) {
 				try {
 					Future<String> future =
 							RedisLockRegistry.this.unlockNotifyMessageListener.subscribeLock(this.lockKey);
 					//DCL
-					if (!obtainLock()) {
-						try {
-							//if short expireAfter key expire for ttl, no receive unlock msg
-							long waitTime = time >= 0 ? time : RedisLockRegistry.this.expireAfter;
-							future.get(waitTime, TimeUnit.MILLISECONDS);
-						}
-						catch (TimeoutException ignore) {
-						}
-						return obtainLock();
+					if (obtainLock()) {
+						return true;
+					}
+					try {
+						//if short expireAfter key expire for ttl, no receive unlock msg
+						long waitTime = time >= 0 ? time : RedisLockRegistry.this.expireAfter;
+						future.get(waitTime, TimeUnit.MILLISECONDS);
+					}
+					catch (TimeoutException ignore) {
+					}
+					if (obtainLock()) {
+						return true;
 					}
 				}
 				finally {
 					RedisLockRegistry.this.unlockNotifyMessageListener.unSubscribeLock(this.lockKey);
 				}
 			}
-			return true;
+			return false;
 		}
 
 		private boolean obtainLock() {

spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 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.
@@ -692,6 +692,60 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
 	}
 
 
+	@Test
+	@RedisAvailable
+	public void earlyWakeUpTest() throws InterruptedException {
+		final int THREAD_CNT = 2;
+		final String testKey = "testKey";
+
+		final CountDownLatch tryLockReady = new CountDownLatch(THREAD_CNT);
+		final CountDownLatch awaitTimeout = new CountDownLatch(THREAD_CNT);
+		final RedisConnectionFactory connectionFactory = getConnectionFactoryForTest();
+		final RedisLockRegistry registry1 = new RedisLockRegistry(connectionFactory, this.registryKey);
+		final RedisLockRegistry registry2 = new RedisLockRegistry(connectionFactory, this.registryKey);
+		final RedisLockRegistry registry3 = new RedisLockRegistry(connectionFactory, this.registryKey);
+		final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_CNT);
+
+		Lock lock1 = registry1.obtain(testKey);
+		Lock lock2 = registry2.obtain(testKey);
+		Lock lock3 = registry3.obtain(testKey);
+		AtomicInteger expectOne = new AtomicInteger();
+
+		lock1.lock();
+		executorService.submit(() -> {
+			try {
+				tryLockReady.countDown();
+				boolean b = lock2.tryLock(10, TimeUnit.SECONDS);
+				awaitTimeout.countDown();
+				if (b) {
+					expectOne.incrementAndGet();
+				}
+			}
+			catch (InterruptedException ignore) {
+			}
+		});
+
+		executorService.submit(() -> {
+			try {
+				tryLockReady.countDown();
+				boolean b = lock3.tryLock(10, TimeUnit.SECONDS);
+				awaitTimeout.countDown();
+				if (b) {
+					expectOne.incrementAndGet();
+				}
+			}
+			catch (InterruptedException ignore) {
+			}
+		});
+
+		assertThat(tryLockReady.await(10, TimeUnit.SECONDS)).isTrue();
+		lock1.unlock();
+		assertThat(awaitTimeout.await(1, TimeUnit.SECONDS)).isFalse();
+		assertThat(expectOne.get()).isEqualTo(1);
+		executorService.shutdown();
+	}
+
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testUlink() {