INT-4246: Revoke leader when lock isn't acquired

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

The `LockRegistryLeaderInitiator` currently does not revoke leadership
when leading instance is unable to acquire lock from the underlying lock.
This can result in multiple `LockRegistryLeaderInitiator` instances becoming
 leaders in the exceptional situations such as lock timeouts.

* Handle leadership revoking when leading `LockRegistryLeaderInitiator`
is unable to acquire lock

Polish javadoc

**Cherry-pick to master**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -55,6 +55,7 @@ import org.springframework.util.Assert;
  *
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Vedran Pavic
  * @since 4.3.1
  */
 public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware {
@@ -323,11 +324,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							if (acquired) {
 								// Success: we are now leader
 								this.locked = true;
-								LockRegistryLeaderInitiator.this.candidate.onGranted(this.context);
-								if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
-									LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted(
-											LockRegistryLeaderInitiator.this, this.context, this.lockKey);
-								}
+								handleGranted();
 							}
 						}
 						else if (acquired) {
@@ -338,6 +335,9 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							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);
 						}
@@ -347,12 +347,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 							this.lock.unlock();
 							this.locked = false;
 							// The lock was broken and we are no longer leader
-							LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context);
-							if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
-								LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked(
-										LockRegistryLeaderInitiator.this, this.context,
-										LockRegistryLeaderInitiator.this.candidate.getRole());
-							}
+							handleRevoked();
 							// Give it a chance to elect some other leader.
 							Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
 							Thread.currentThread().interrupt();
@@ -365,12 +360,7 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 				this.lock.unlock();
 				if (this.locked) {
 					// We are stopping, therefore not leading any more
-					LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context);
-					if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
-						LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked(
-								LockRegistryLeaderInitiator.this, this.context,
-								LockRegistryLeaderInitiator.this.candidate.getRole());
-					}
+					handleRevoked();
 				}
 				this.locked = false;
 			}
@@ -381,6 +371,23 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 			return this.locked;
 		}
 
+		private void handleGranted() throws InterruptedException {
+			LockRegistryLeaderInitiator.this.candidate.onGranted(this.context);
+			if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
+				LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted(
+						LockRegistryLeaderInitiator.this, this.context, this.lockKey);
+			}
+		}
+
+		private void handleRevoked() {
+			LockRegistryLeaderInitiator.this.candidate.onRevoked(this.context);
+			if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
+				LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked(
+						LockRegistryLeaderInitiator.this, this.context,
+						LockRegistryLeaderInitiator.this.candidate.getRole());
+			}
+		}
+
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2016 the original author or authors.
+ * Copyright 2012-2017 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.
@@ -18,9 +18,16 @@ package org.springframework.integration.support.leader;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
 
 import org.apache.log4j.Level;
 import org.junit.Before;
@@ -37,6 +44,7 @@ import org.springframework.integration.test.rule.Log4jLevelAdjuster;
 /**
  * @author Dave Syer
  * @author Artem Bilan
+ * @author Vedran Pavic
  *
  * @since 4.3.1
  */
@@ -100,6 +108,64 @@ public class LockRegistryLeaderInitiatorTests {
 		assertThat(another.getContext().isLeader(), is(true));
 	}
 
+	@Test
+	public void competingWithLock() throws Exception {
+		// switch used to toggle which registry obtains lock
+		AtomicBoolean firstLocked = new AtomicBoolean(true);
+
+		// set up first registry instance - this one will be able to obtain lock initially
+		LockRegistry firstRegistry = mock(LockRegistry.class);
+		Lock firstLock = mock(Lock.class);
+		given(firstRegistry.obtain(anyString())).willReturn(firstLock);
+		given(firstLock.tryLock(anyLong(), any(TimeUnit.class))).willAnswer(i -> firstLocked.get());
+
+		// set up first initiator instance using first LockRegistry
+		LockRegistryLeaderInitiator first =
+				new LockRegistryLeaderInitiator(firstRegistry, new DefaultCandidate());
+		CountDownLatch firstGranted = new CountDownLatch(1);
+		CountDownLatch firstRevoked = new CountDownLatch(1);
+		first.setHeartBeatMillis(10);
+		first.setBusyWaitMillis(1);
+		first.setLeaderEventPublisher(new CountingPublisher(firstGranted, firstRevoked));
+
+		// set up second registry instance - this one will NOT be able to obtain lock initially
+		LockRegistry secondRegistry = mock(LockRegistry.class);
+		Lock secondLock = mock(Lock.class);
+		given(secondRegistry.obtain(anyString())).willReturn(secondLock);
+		given(secondLock.tryLock(anyLong(), any(TimeUnit.class))).willAnswer(i -> !firstLocked.get());
+
+		// set up second initiator instance using second LockRegistry
+		LockRegistryLeaderInitiator second =
+				new LockRegistryLeaderInitiator(secondRegistry, new DefaultCandidate());
+		CountDownLatch secondGranted = new CountDownLatch(1);
+		CountDownLatch secondRevoked = new CountDownLatch(1);
+		second.setHeartBeatMillis(10);
+		second.setBusyWaitMillis(1);
+		second.setLeaderEventPublisher(new CountingPublisher(secondGranted, secondRevoked));
+
+		// start initiators
+		first.start();
+		second.start();
+
+		Thread.sleep(100);
+
+		// first initiator should lead and publish granted event
+		assertThat(first.getContext().isLeader(), is(true));
+		assertThat(second.getContext().isLeader(), is(false));
+		assertThat(firstGranted.await(10, TimeUnit.SECONDS), is(true));
+
+		// simulate first registry instance unable to obtain lock, for example due to lock timeout
+		firstLocked.set(false);
+
+		Thread.sleep(100);
+
+		// second initiator should take lead and publish granted event, first initiator should publish revoked event
+		assertThat(second.getContext().isLeader(), is(true));
+		assertThat(first.getContext().isLeader(), is(false));
+		assertThat(secondGranted.await(10, TimeUnit.SECONDS), is(true));
+		assertThat(firstRevoked.await(10, TimeUnit.SECONDS), is(true));
+	}
+
 	private static class CountingPublisher implements LeaderEventPublisher {
 
 		private final CountDownLatch granted;

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java

@@ -44,7 +44,7 @@ import org.springframework.util.Assert;
  *
  * @author Dave Syer
  * @author Artem Bilan
- * @author Vedran Pavić
+ * @author Vedran Pavic
  *
  * @since 4.3
  */