Fix one more race condition in RedisLeaderTests

https://build.spring.io/browse/INT-SI50X-JOB1-56

We can't wait for the latch in the interruptable code flow;
we can't have a round-robing election guarantees.

* Add `Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis)`
to the `LockRegistryLeaderInitiator` when we restart the main task
* Remove latches waiting and thread shifting from the
`RedisLockRegistryLeaderInitiatorTests`
* Use long `busyWaitMillis` for yielding initiator to let the second
candidate to be elected

**Cherry-pick to 5.0.x**

(cherry picked from commit 081d0d1)

Author: artembilan

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

@@ -408,7 +408,12 @@ else if (acquired) {
 							if (isRunning()) {
 								logger.warn("Restarting LeaderSelector for " + this.context + " because of error.", e);
 								LockRegistryLeaderInitiator.this.future =
-										LockRegistryLeaderInitiator.this.executorService.submit(this);
+										LockRegistryLeaderInitiator.this.executorService.submit(
+												() -> {
+													// Give it a chance to elect some other leader.
+													Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
+													return call();
+												});
 							}
 							return null;
 						}

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

@@ -23,12 +23,9 @@
 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;
 
@@ -40,8 +37,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.integration.test.support.LongRunningIntegrationTest;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
-import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Artem Bilan
@@ -52,7 +49,8 @@
  */
 public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
 
-	private static final Log logger = LogFactory.getLog(RedisLockRegistryLeaderInitiatorTests.class);
+	@Rule
+	public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
 
 	@Rule
 	public Log4j2LevelAdjuster adjuster =
@@ -105,63 +103,37 @@ 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) {
+		initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1, acquireLockFailed1));
 
-			@Override
-			public void publishOnRevoked(Object source, Context context, String role) {
-				latchesExecutor.execute(() -> {
-					// It's difficult to see round-robin election, so block one initiator until the second is elected.
-					try {
-						assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
-					}
-					catch (InterruptedException e) {
-						ReflectionUtils.rethrowRuntimeException(e);
-					}
-					super.publishOnRevoked(source, context, role);
-				});
+		initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2, acquireLockFailed2));
 
-			}
-
-		});
-
-		initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2, acquireLockFailed2) {
-
-			@Override
-			public void publishOnRevoked(Object source, Context context, String 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);
-				});
-			}
-
-		});
+		// It's hard to see round-robin election, so let's make the yielding initiator to sleep long before restarting
+		initiator1.setBusyWaitMillis(5000);
 
 		initiator1.getContext().yield();
 
 		assertThat(revoked1.await(10, TimeUnit.SECONDS), is(true));
+		assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
 
 		assertThat(initiator2.getContext().isLeader(), is(true));
 		assertThat(initiator1.getContext().isLeader(), is(false));
 
+		initiator1.setBusyWaitMillis(LockRegistryLeaderInitiator.DEFAULT_BUSY_WAIT_TIME);
+		initiator2.setBusyWaitMillis(5000);
+
 		initiator2.getContext().yield();
 
 		assertThat(revoked2.await(10, TimeUnit.SECONDS), is(true));
+		assertThat(granted1.await(10, TimeUnit.SECONDS), is(true));
 
 		assertThat(initiator1.getContext().isLeader(), is(true));
 		assertThat(initiator2.getContext().isLeader(), is(false));
 
 		initiator2.stop();
 
 		CountDownLatch revoked11 = new CountDownLatch(1);
-		initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11, new CountDownLatch(1)));
+		initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11,
+				new CountDownLatch(1)));
 
 		initiator1.getContext().yield();