Fix RedisLockRegLeaderInitTests race condition

https://build.spring.io/browse/INT-FATS5IC-501/

When we try to wait for the `Latch` in the interruptable code flow,
it is a fact that we step away from the waiting and end up with the
race condition downstream.

* Wrap `Latch` in the interruptable `publishOnRevoked()` code to the
`Executor.execute()`
* Remove `deleteTimeoutMillis` option from the `RedisLockRegistry`
since it doesn't make sense in the interruptable code.
* Add `RedisLockRegistry.setExecutor()` to allow to inject an external
`Executor`
* Add more debug logging into the `LockRegistryLeaderInitiator`

**Cherry-pick to 5.0.x**

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

@@ -23,6 +23,7 @@ import java.util.Iterator;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -79,8 +80,6 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 
 	private static final long DEFAULT_EXPIRE_AFTER = 60000L;
 
-	public static final long DEFAULT_DELETE_TIMEOUT = 10000L;
-
 	private static final String OBTAIN_LOCK_SCRIPT =
 			"local lockClientId = redis.call('GET', KEYS[1])\n" +
 					"if lockClientId == ARGV[1] then\n" +
@@ -97,21 +96,14 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 	 * An {@link ExecutorService} to call {@link StringRedisTemplate#delete(Object)} in
 	 * the separate thread when the current one is interrupted.
 	 */
-	private ExecutorService executorService =
+	private Executor executor =
 			Executors.newCachedThreadPool(new CustomizableThreadFactory("redis-lock-registry-"));
 
 	/**
 	 * 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 executorServiceExplicitlySet;
-
-	/**
-	 * Time in milliseconds to wait for the {@link StringRedisTemplate#delete(Object)} in
-	 * the background thread.
-	 */
-	private long deleteTimeoutMillis = DEFAULT_DELETE_TIMEOUT;
-
+	private boolean executorExplicitlySet;
 
 	private final Map<String, RedisLock> locks = new ConcurrentHashMap<>();
 
@@ -149,6 +141,17 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 		this.expireAfter = expireAfter;
 	}
 
+	/**
+	 * Set the {@link Executor}, where is not provided then a default of
+	 * cached thread pool Executor will be used.
+	 * @param executor the executor service
+	 * @since 5.0.5
+	 */
+	public void setExecutor(Executor executor) {
+		this.executor = executor;
+		this.executorExplicitlySet = true;
+	}
+
 	@Override
 	public Lock obtain(Object lockKey) {
 		Assert.isInstanceOf(String.class, lockKey);
@@ -171,8 +174,8 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 
 	@Override
 	public void destroy() {
-		if (!this.executorServiceExplicitlySet) {
-			this.executorService.shutdown();
+		if (!this.executorExplicitlySet) {
+			((ExecutorService) this.executor).shutdown();
 		}
 	}
 
@@ -298,11 +301,9 @@ public final class RedisLockRegistry implements ExpirableLockRegistry, Disposabl
 				return;
 			}
 			try {
-
 				if (Thread.currentThread().isInterrupted()) {
-					RedisLockRegistry.this.executorService.submit(() ->
-							RedisLockRegistry.this.redisTemplate.delete(this.lockKey))
-							.get(RedisLockRegistry.this.deleteTimeoutMillis, TimeUnit.MILLISECONDS);
+					RedisLockRegistry.this.executor.execute(() ->
+							RedisLockRegistry.this.redisTemplate.delete(this.lockKey));
 				}
 				else {
 					RedisLockRegistry.this.redisTemplate.delete(this.lockKey);

spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java

@@ -23,8 +23,12 @@ import static org.junit.Assert.assertThat;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.junit.Rule;
 import org.junit.Test;
 
@@ -36,6 +40,8 @@ import org.springframework.integration.redis.rules.RedisAvailableTests;
 import org.springframework.integration.redis.util.RedisLockRegistry;
 import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
 import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Artem Bilan
@@ -46,10 +52,12 @@ import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
  */
 public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 
+	private static final Log logger = LogFactory.getLog(RedisLockRegistryLeaderInitiatorTests.class);
+
 	@Rule
 	public Log4j2LevelAdjuster adjuster =
 			Log4j2LevelAdjuster.trace()
-					.categories(true, "org.springframework.data.redis");
+					.categories(true, "org.springframework.integration.redis.leader");
 
 	@Test
 	@RedisAvailable
@@ -62,6 +70,8 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 		for (int i = 0; i < 2; i++) {
 			LockRegistryLeaderInitiator initiator =
 					new LockRegistryLeaderInitiator(registry, new DefaultCandidate("foo:" + i, "bar"));
+			initiator.setExecutorService(
+					Executors.newSingleThreadExecutor(new CustomizableThreadFactory("lock-leadership-" + i + "-")));
 			initiator.setLeaderEventPublisher(countingPublisher);
 			initiators.add(initiator);
 		}
@@ -95,18 +105,23 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 		CountDownLatch acquireLockFailed1 = new CountDownLatch(1);
 		CountDownLatch acquireLockFailed2 = new CountDownLatch(1);
 
+		Executor latchesExecutor = Executors.newCachedThreadPool();
+
 		initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1, acquireLockFailed1) {
 
 			@Override
 			public void publishOnRevoked(Object source, Context context, String role) {
-				try {
+				latchesExecutor.execute(() -> {
 					// It's difficult to see round-robin election, so block one initiator until the second is elected.
-					assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
-				}
-				catch (InterruptedException e) {
-					// No op
-				}
-				super.publishOnRevoked(source, context, role);
+					try {
+						assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
+					}
+					catch (InterruptedException e) {
+						ReflectionUtils.rethrowRuntimeException(e);
+					}
+					super.publishOnRevoked(source, context, role);
+				});
+
 			}
 
 		});
@@ -115,14 +130,16 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 
 			@Override
 			public void publishOnRevoked(Object source, Context context, String role) {
-				try {
-					// It's difficult to see round-robin election, so block one initiator until the second is elected.
-					assertThat(granted1.await(10, TimeUnit.SECONDS), is(true));
-				}
-				catch (InterruptedException e) {
-					// No op
-				}
-				super.publishOnRevoked(source, context, role);
+				latchesExecutor.execute(() -> {
+					try {
+						// It's difficult to see round-robin election, so block one initiator until the second is elected.
+						assertThat(granted1.await(10, TimeUnit.SECONDS), is(true));
+					}
+					catch (InterruptedException e) {
+						ReflectionUtils.rethrowRuntimeException(e);
+					}
+					super.publishOnRevoked(source, context, role);
+				});
 			}
 
 		});