INT-4254: LockRegLInit: Fix unconditional unlock

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

The `LockRegistryLeaderInitiator#LeaderSelector` unconditional calls
`unlock()` in the `finally` block when the `Lock` might not be locked.

Another problem that `running = true` is set after submitting `LeaderSelector` task.
That might bring the problem that `LeaderSelector` won't be selected because of
`this.running` race condition

* Move `this.running = true` before submitting `LeaderSelector` task
* Move `unlock()` into the `if (this.locked)` condition

**Cherry-pick to 4.3.x**

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

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

@@ -253,8 +253,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 		synchronized (this.lifecycleMonitor) {
 			if (!this.running) {
 				this.leaderSelector = new LeaderSelector(buildLeaderPath());
-				this.future = this.executorService.submit(this.leaderSelector);
 				this.running = true;
+				this.future = this.executorService.submit(this.leaderSelector);
 				logger.debug("Started LeaderInitiator");
 			}
 		}
@@ -357,8 +357,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 				}
 			}
 			finally {
-				this.lock.unlock();
 				if (this.locked) {
+					this.lock.unlock();
 					// We are stopping, therefore not leading any more
 					handleRevoked();
 				}

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

@@ -17,22 +17,31 @@
 package org.springframework.integration.support.leader;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.support.ExecutorServiceAdapter;
 import org.springframework.integration.leader.Context;
 import org.springframework.integration.leader.DefaultCandidate;
 import org.springframework.integration.leader.event.LeaderEventPublisher;
@@ -52,9 +61,9 @@ public class LockRegistryLeaderInitiatorTests {
 
 	private CountDownLatch revoked;
 
-	private LockRegistry registry = new DefaultLockRegistry();
+	private final LockRegistry registry = new DefaultLockRegistry();
 
-	private LockRegistryLeaderInitiator initiator =
+	private final LockRegistryLeaderInitiator initiator =
 			new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate());
 
 	@Before
@@ -162,6 +171,49 @@ public class LockRegistryLeaderInitiatorTests {
 		second.stop();
 	}
 
+	@Test
+	public void testGracefulLeaderSelectorExit() throws Exception {
+		AtomicReference<Throwable> throwableAtomicReference = new AtomicReference<>();
+
+		LockRegistry registry = mock(LockRegistry.class);
+
+		Lock lock = spy(new ReentrantLock());
+
+		willAnswer(invocation -> {
+			try {
+				return invocation.callRealMethod();
+			}
+			catch (Throwable e) {
+				throwableAtomicReference.set(e);
+				throw e;
+			}
+		})
+				.given(lock)
+				.unlock();
+
+
+		given(registry.obtain(anyString()))
+				.willReturn(lock);
+
+		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry);
+
+		willAnswer(invocation -> {
+			initiator.stop();
+			return false;
+		})
+				.given(lock)
+				.tryLock(anyLong(), eq(TimeUnit.MILLISECONDS));
+
+		new DirectFieldAccessor(initiator).setPropertyValue("executorService",
+				new ExecutorServiceAdapter(
+						new SyncTaskExecutor()));
+
+		initiator.start();
+
+		Throwable throwable = throwableAtomicReference.get();
+		assertNull(throwable);
+	}
+
 	private static class CountingPublisher implements LeaderEventPublisher {
 
 		private final CountDownLatch granted;