INT-4396: Add retrying lock in case of exception

JIRA: https://jira.spring.io/browse/INT-4396

When target distributed `Lock` implementation throws an exception, e.g.
in case of no connection to the service, the `LockRegistryLeaderInitiator`
exists the loop and can come back to the elections only after restart

* Catch all the exception on `this.lock.tryLock()` and resubmit
`LeaderSelector` for a new locking cycle if
`LockRegistryLeaderInitiator.isRunning()` and `InterruptedException`

* Remove diagnostics from the `JdbcLockRegistryLeaderInitiatorTests`
since this fix confirms that we just didn't have a reconnect logic before
when this test failed sporadically

**Cherry-pick to 4.3.x**

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -326,7 +326,7 @@ protected class LeaderSelector implements Callable<Void> {
 		@Override
 		public Void call() throws Exception {
 			try {
-				while (LockRegistryLeaderInitiator.this.running) {
+				while (isRunning()) {
 					try {
 						// We always try to acquire the lock, in case it expired
 						boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
@@ -345,26 +345,40 @@ else if (acquired) {
 							// If we were able to acquire it but we were already locked we
 							// should release it
 							this.lock.unlock();
-							// Give it a chance to expire.
-							Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis);
+							if (isRunning()) {
+								// Give it a chance to expire.
+								Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis);
+							}
 						}
 						else {
 							this.locked = false;
 							// We were not able to acquire it, therefore not leading any more
 							handleRevoked();
-							// Try again quickly in case the lock holder dropped it
-							Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
+							if (isRunning()) {
+								// Try again quickly in case the lock holder dropped it
+								Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
+							}
 						}
 					}
-					catch (InterruptedException e) {
+					catch (Exception e) {
 						if (this.locked) {
 							this.lock.unlock();
 							this.locked = false;
 							// The lock was broken and we are no longer leader
 							handleRevoked();
-							// Give it a chance to elect some other leader.
-							Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
+							if (isRunning()) {
+								// Give it a chance to elect some other leader.
+								Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
+							}
+						}
+
+						if (e instanceof InterruptedException) {
 							Thread.currentThread().interrupt();
+							if (isRunning()) {
+								logger.warn("Restarting LeaderSelector because of error.", e);
+								LockRegistryLeaderInitiator.this.future =
+										LockRegistryLeaderInitiator.this.executorService.submit(this);
+							}
 							return null;
 						}
 					}
@@ -446,11 +460,6 @@ public boolean isLeader() {
 		public void yield() {
 			if (LockRegistryLeaderInitiator.this.future != null) {
 				LockRegistryLeaderInitiator.this.future.cancel(true);
-				if (isRunning()) {
-					LockRegistryLeaderInitiator.this.future =
-							LockRegistryLeaderInitiator.this.executorService
-									.submit(LockRegistryLeaderInitiator.this.leaderSelector);
-				}
 			}
 		}
 

spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2017 the original author or authors.
+ * Copyright 2012-2018 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.
@@ -265,6 +265,38 @@ public void testGracefulLeaderSelectorExit() throws Exception {
 		assertNull(throwable);
 	}
 
+	@Test
+	public void testExceptionFromLock() throws Exception {
+		Lock mockLock = mock(Lock.class);
+
+		AtomicBoolean exceptionThrown = new AtomicBoolean();
+
+		willAnswer(invocation -> {
+			if (!exceptionThrown.getAndSet(true)) {
+				throw new RuntimeException("lock is broken");
+			}
+			else {
+				return true;
+			}
+		}).given(mockLock).tryLock(anyLong(), any(TimeUnit.class));
+
+		LockRegistry registry = lockKey -> mockLock;
+
+		CountDownLatch onGranted = new CountDownLatch(1);
+
+		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry);
+
+		initiator.setLeaderEventPublisher(new CountingPublisher(onGranted));
+
+		initiator.start();
+
+		assertTrue(onGranted.await(10, TimeUnit.SECONDS));
+		assertTrue(initiator.getContext().isLeader());
+		assertTrue(exceptionThrown.get());
+
+		initiator.stop();
+	}
+
 	private static class CountingPublisher implements LeaderEventPublisher {
 
 		private final CountDownLatch granted;

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java

@@ -27,7 +27,6 @@
 
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 
 import org.springframework.integration.jdbc.lock.DefaultLockRepository;
@@ -36,7 +35,6 @@
 import org.springframework.integration.leader.DefaultCandidate;
 import org.springframework.integration.leader.event.LeaderEventPublisher;
 import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
-import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
@@ -52,14 +50,6 @@ public class JdbcLockRegistryLeaderInitiatorTests {
 
 	public static EmbeddedDatabase dataSource;
 
-	@Rule
-	public Log4j2LevelAdjuster adjuster =
-			Log4j2LevelAdjuster.trace()
-					.categories("org.springframework.integration",
-							"org.springframework.integration.jdbc",
-							"org.springframework.jdbc",
-							"org.apache.derby");
-
 	@BeforeClass
 	public static void init() {
 		dataSource = new EmbeddedDatabaseBuilder()