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**

(cherry picked from commit 51c4951)

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java

@@ -315,7 +315,7 @@ public void stop() {
 					this.future.cancel(true);
 				}
 				this.future = null;
-				logger.debug("Stopped LeaderInitiator");
+				logger.debug("Stopped LeaderInitiator for " + getContext());
 			}
 		}
 	}
@@ -347,6 +347,11 @@ public Void call() throws Exception {
 			try {
 				while (isRunning()) {
 					try {
+
+						if (logger.isDebugEnabled()) {
+							logger.debug("Acquiring the lock for " + this.context);
+						}
+
 						// We always try to acquire the lock, in case it expired
 						boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
 								TimeUnit.MILLISECONDS);
@@ -386,8 +391,8 @@ else if (acquired) {
 								this.lock.unlock();
 							}
 							catch (Exception e1) {
-								logger.debug("Could not unlock - treat as broken: " + this.context +
-										". Revoking " + (isRunning() ? " and retrying..." : "..."), e);
+								logger.debug("Could not unlock - treat as broken " + this.context +
+										". Revoking " + (isRunning() ? " and retrying..." : "..."), e1);
 
 							}
 							// The lock was broken and we are no longer leader
@@ -401,14 +406,15 @@ else if (acquired) {
 						if (e instanceof InterruptedException || Thread.currentThread().isInterrupted()) {
 							Thread.currentThread().interrupt();
 							if (isRunning()) {
-								logger.warn("Restarting LeaderSelector because of error.", e);
+								logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e);
 								LockRegistryLeaderInitiator.this.future =
 										LockRegistryLeaderInitiator.this.executorService.submit(this);
 							}
 							return null;
 						}
 						else if (logger.isDebugEnabled()) {
-							logger.debug("Error acquiring the lock. " + (isRunning() ? "Retrying..." : ""), e);
+							logger.debug("Error acquiring the lock for " + this.context +
+									". " + (isRunning() ? "Retrying..." : ""), e);
 						}
 					}
 				}
@@ -420,7 +426,8 @@ else if (logger.isDebugEnabled()) {
 						this.lock.unlock();
 					}
 					catch (Exception e) {
-						logger.debug("Could not unlock during stop - treat as broken. Revoking...", e);
+						logger.debug("Could not unlock during stop for " + this.context
+								+ " - treat as broken. Revoking...", e);
 					}
 					// We are stopping, therefore not leading any more
 					handleRevoked();
@@ -492,6 +499,9 @@ public boolean isLeader() {
 
 		@Override
 		public void yield() {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Yielding leadership from " + this);
+			}
 			if (LockRegistryLeaderInitiator.this.future != null) {
 				LockRegistryLeaderInitiator.this.future.cancel(true);
 			}

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

@@ -23,6 +23,7 @@
 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 RedisLockRegistry(RedisConnectionFactory connectionFactory, String regist
 		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 void expireUnusedOlderThan(long age) {
 
 	@Override
 	public void destroy() {
-		if (!this.executorServiceExplicitlySet) {
-			this.executorService.shutdown();
+		if (!this.executorExplicitlySet) {
+			((ExecutorService) this.executor).shutdown();
 		}
 	}
 
@@ -298,11 +301,9 @@ public void unlock() {
 				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 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.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 @@
  */
 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 void testDistributedLeaderElection() throws Exception {
 		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 void testDistributedLeaderElection() throws Exception {
 		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 void publishOnRevoked(Object source, Context context, String role) {
 
 			@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);
+				});
 			}
 
 		});