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

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java

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.
@@ -312,7 +312,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
 						// TODO obtain(this.lockKey) because of INT-4248. Should be fixed in 5.0
@@ -330,27 +330,41 @@ else if (acquired) {
 							// should release it
 							LockRegistryLeaderInitiator.this.locks.obtain(this.lockKey)
 									.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) {
 							LockRegistryLeaderInitiator.this.locks.obtain(this.lockKey)
 									.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;
 						}
 					}
@@ -405,11 +419,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.
@@ -214,6 +214,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

@@ -28,7 +28,6 @@
 import org.apache.log4j.Level;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 
 import org.springframework.integration.jdbc.lock.DefaultLockRepository;
@@ -37,7 +36,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.Log4jLevelAdjuster;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
 import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
@@ -52,10 +50,6 @@ public class JdbcLockRegistryLeaderInitiatorTests {
 
 	public static EmbeddedDatabase dataSource;
 
-	@Rule
-	public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration",
-			"org.springframework.integration.jdbc", "org.springframework.jdbc", "org.apache.derby");
-
 	@BeforeClass
 	public static void init() {
 		dataSource = new EmbeddedDatabaseBuilder()