INT-4447: LockRegLeaderInit: Catch unlock errors

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

When we get an exception during `this.lock.unlock()`, we don't revoke
leadership.
In case of external resource (e.g. JDBC) this may cause a race condition
when the second candidate is selected as leader when connection comes back

* Catch `this.lock.unlock()` exceptions and log them under DEBUG.
This way we proceed to the `handleRevoked()` logic

**Cherry-pick to 5.0.x and 4.3.x**

Author: artembilan

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

@@ -381,8 +381,15 @@ else if (acquired) {
 					}
 					catch (Exception e) {
 						if (this.locked) {
-							this.lock.unlock();
 							this.locked = false;
+							try {
+								this.lock.unlock();
+							}
+							catch (Exception e1) {
+								logger.debug("Could not unlock - treat as broken. " +
+										"Revoking " + (isRunning() ? " and retrying..." : "..."), e);
+
+							}
 							// The lock was broken and we are no longer leader
 							handleRevoked();
 							if (isRunning()) {
@@ -408,11 +415,16 @@ else if (logger.isDebugEnabled()) {
 			}
 			finally {
 				if (this.locked) {
-					this.lock.unlock();
+					this.locked = false;
+					try {
+						this.lock.unlock();
+					}
+					catch (Exception e) {
+						logger.debug("Could not unlock during stop - treat as broken. Revoking...", e);
+					}
 					// We are stopping, therefore not leading any more
 					handleRevoked();
 				}
-				this.locked = false;
 			}
 			return null;
 		}

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

@@ -167,6 +167,35 @@ public void publishOnRevoked(Object source, Context context, String role) {
 		initiator1.stop();
 	}
 
+	@Test
+	public void testLostConnection() throws InterruptedException {
+		CountDownLatch granted = new CountDownLatch(1);
+		CountingPublisher countingPublisher = new CountingPublisher(granted);
+
+		DefaultLockRepository lockRepository = new DefaultLockRepository(dataSource);
+		lockRepository.afterPropertiesSet();
+		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(new JdbcLockRegistry(lockRepository));
+		initiator.setLeaderEventPublisher(countingPublisher);
+
+		initiator.start();
+
+		assertThat(granted.await(10, TimeUnit.SECONDS), is(true));
+
+		destroy();
+
+		assertThat(countingPublisher.revoked.await(10, TimeUnit.SECONDS), is(true));
+
+		granted = new CountDownLatch(1);
+		countingPublisher = new CountingPublisher(granted);
+		initiator.setLeaderEventPublisher(countingPublisher);
+
+		init();
+
+		assertThat(granted.await(10, TimeUnit.SECONDS), is(true));
+
+		initiator.stop();
+	}
+
 	private static class CountingPublisher implements LeaderEventPublisher {
 
 		private final CountDownLatch granted;