GH-2354 Injection Executor to LockLeaderInitiator

Fixes #2354

INT-2354 missed file in previous commit

INT-2354 fix failing test

INT-2354 changes following feedback

INT-2354 feedback changes

INT-2354 feedback changes

* Simple code style polishing

Author: boaty82

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

@@ -57,6 +57,7 @@
  * @author Artem Bilan
  * @author Vedran Pavic
  * @author Glenn Renfro
+ * @author Kiel Boatman
  *
  * @since 4.3.1
  */
@@ -72,12 +73,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 
 	private final Object lifecycleMonitor = new Object();
 
-	/**
-	 * Executor service for running leadership daemon.
-	 */
-	private final ExecutorService executorService =
-			Executors.newSingleThreadExecutor(new CustomizableThreadFactory("lock-leadership-"));
-
 	/**
 	 * A lock registry. The locks it manages should be global (whatever that means for the
 	 * system) and expiring, in case the holder dies without notifying anyone.
@@ -92,6 +87,17 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 	 */
 	private final Candidate candidate;
 
+	/**
+	 * Executor service for running leadership daemon.
+	 */
+	private ExecutorService executorService =
+			Executors.newSingleThreadExecutor(new CustomizableThreadFactory("lock-leadership-"));
+
+	/**
+	 * Flag to denote whether the {@link ExecutorService} was provided via the setter and
+	 * thus should not be shutdown when {@link #destroy()} is called
+	 */
+	private boolean executorServiceExplicitlySet;
 
 	/**
 	 * Time in milliseconds to wait in between attempts to re-acquire the lock, once it is
@@ -124,28 +130,28 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
 	 */
 	private LeaderEventPublisher leaderEventPublisher;
 
-	/**
-	 * Future returned by submitting an {@link LeaderSelector} to
-	 * {@link #executorService}. This is used to cancel leadership.
-	 */
-	private volatile Future<?> future;
-
 	/**
 	 * @see SmartLifecycle
 	 */
-	private volatile boolean autoStartup = true;
+	private boolean autoStartup = true;
 
 	/**
 	 * @see SmartLifecycle which is an extension of org.springframework.context.Phased
 	 */
-	private volatile int phase;
+	private int phase;
 
 	/**
 	 * Flag that indicates whether the leadership election for this {@link #candidate} is
 	 * running.
 	 */
 	private volatile boolean running;
 
+	/**
+	 * Future returned by submitting an {@link LeaderSelector} to
+	 * {@link #executorService}. This is used to cancel leadership.
+	 */
+	private volatile Future<?> future;
+
 	/**
 	 * Create a new leader initiator with the provided lock registry and a default
 	 * candidate (which just logs the leadership events).
@@ -168,9 +174,15 @@ public LockRegistryLeaderInitiator(LockRegistry locks, Candidate candidate) {
 		this.candidate = candidate;
 	}
 
-	@Override
-	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
-		this.applicationEventPublisher = applicationEventPublisher;
+	/**
+	 * Set the {@link ExecutorService}, where is not provided then a default of
+	 * single thread Executor will be used.
+	 * @param executorService the executor service
+	 * @since 5.0.2
+	 */
+	public void setExecutorService(ExecutorService executorService) {
+		this.executorService = executorService;
+		this.executorServiceExplicitlySet = true;
 	}
 
 	public void setHeartBeatMillis(long heartBeatMillis) {
@@ -182,13 +194,18 @@ public void setBusyWaitMillis(long busyWaitMillis) {
 	}
 
 	/**
-	 * Sets the {@link LeaderEventPublisher}.
+	 * Set the {@link LeaderEventPublisher}.
 	 * @param leaderEventPublisher the event publisher
 	 */
 	public void setLeaderEventPublisher(LeaderEventPublisher leaderEventPublisher) {
 		this.leaderEventPublisher = leaderEventPublisher;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 	/**
 	 * @return true if leadership election for this {@link #candidate} is running.
 	 */
@@ -272,9 +289,11 @@ public void start() {
 	}
 
 	@Override
-	public void destroy() throws Exception {
+	public void destroy() {
 		stop();
-		this.executorService.shutdown();
+		if (!this.executorServiceExplicitlySet) {
+			this.executorService.shutdown();
+		}
 	}
 
 	@Override

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.support.leader;
 
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -30,6 +31,8 @@
 import static org.mockito.Mockito.spy;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -48,20 +51,22 @@
 import org.springframework.integration.leader.event.LeaderEventPublisher;
 import org.springframework.integration.support.locks.DefaultLockRegistry;
 import org.springframework.integration.support.locks.LockRegistry;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Dave Syer
  * @author Artem Bilan
  * @author Vedran Pavic
  * @author Glenn Renfro
+ * @author Kiel Boatman
  *
  * @since 4.3.1
  */
 public class LockRegistryLeaderInitiatorTests {
 
-	private CountDownLatch granted;
+	private CountDownLatch granted = new CountDownLatch(1);
 
-	private CountDownLatch revoked;
+	private CountDownLatch revoked = new CountDownLatch(1);
 
 	private final LockRegistry registry = new DefaultLockRegistry();
 
@@ -70,8 +75,6 @@ public class LockRegistryLeaderInitiatorTests {
 
 	@Before
 	public void init() {
-		this.granted = new CountDownLatch(1);
-		this.revoked = new CountDownLatch(1);
 		this.initiator.setLeaderEventPublisher(new CountingPublisher(this.granted, this.revoked));
 	}
 
@@ -105,6 +108,7 @@ public void yield() throws Exception {
 	public void competing() throws Exception {
 		LockRegistryLeaderInitiator another =
 				new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate());
+
 		CountDownLatch other = new CountDownLatch(1);
 		another.setLeaderEventPublisher(new CountingPublisher(other));
 		this.initiator.start();
@@ -139,9 +143,7 @@ public void competingWithErrorPublish() throws Exception {
 	public void testExceptionFromEvent() throws Exception {
 		CountDownLatch onGranted = new CountDownLatch(1);
 
-		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate());
-
-		initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() {
+		this.initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() {
 
 			@Override
 			public void publishOnGranted(Object source, Context context, String role) {
@@ -155,12 +157,12 @@ public void publishOnGranted(Object source, Context context, String role) {
 
 		});
 
-		initiator.start();
+		this.initiator.start();
 
 		assertTrue(onGranted.await(10, TimeUnit.SECONDS));
 		assertTrue(initiator.getContext().isLeader());
 
-		initiator.stop();
+		this.initiator.stop();
 	}
 
 	@Test
@@ -177,6 +179,7 @@ public void competingWithLock() throws Exception {
 		// 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);
 		CountDownLatch firstAquireLockFailed = new CountDownLatch(1);
@@ -193,6 +196,7 @@ public void competingWithLock() throws Exception {
 		// 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);
 		CountDownLatch secondAquireLockFailed = new CountDownLatch(1);
@@ -246,20 +250,20 @@ public void testGracefulLeaderSelectorExit() throws Exception {
 		given(registry.obtain(anyString()))
 				.willReturn(lock);
 
-		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry);
+		LockRegistryLeaderInitiator another = new LockRegistryLeaderInitiator(registry);
 
 		willAnswer(invocation -> {
-			initiator.stop();
+			another.stop();
 			return false;
 		})
 				.given(lock)
 				.tryLock(anyLong(), eq(TimeUnit.MILLISECONDS));
 
-		new DirectFieldAccessor(initiator).setPropertyValue("executorService",
+		new DirectFieldAccessor(another).setPropertyValue("executorService",
 				new ExecutorServiceAdapter(
 						new SyncTaskExecutor()));
 
-		initiator.start();
+		another.start();
 
 		Throwable throwable = throwableAtomicReference.get();
 		assertNull(throwable);
@@ -284,17 +288,40 @@ public void testExceptionFromLock() throws Exception {
 
 		CountDownLatch onGranted = new CountDownLatch(1);
 
-		LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(registry);
+		LockRegistryLeaderInitiator another = new LockRegistryLeaderInitiator(registry);
 
-		initiator.setLeaderEventPublisher(new CountingPublisher(onGranted));
+		another.setLeaderEventPublisher(new CountingPublisher(onGranted));
 
-		initiator.start();
+		another.start();
 
 		assertTrue(onGranted.await(10, TimeUnit.SECONDS));
-		assertTrue(initiator.getContext().isLeader());
+		assertTrue(another.getContext().isLeader());
 		assertTrue(exceptionThrown.get());
 
-		initiator.stop();
+		another.stop();
+	}
+
+	@Test
+	public void shouldShutdownInternalExecutorService() {
+		this.initiator.start();
+		this.initiator.destroy();
+
+		ExecutorService executorService =
+				TestUtils.getPropertyValue(this.initiator, "executorService", ExecutorService.class);
+
+		assertTrue(executorService.isShutdown());
+	}
+
+	@Test
+	public void doNotShutdownProvidedExecutorService() {
+		LockRegistryLeaderInitiator another = new LockRegistryLeaderInitiator(this.registry);
+		ExecutorService executorService = Executors.newSingleThreadExecutor();
+		another.setExecutorService(executorService);
+
+		another.start();
+		another.destroy();
+
+		assertFalse(executorService.isShutdown());
 	}
 
 	private static class CountingPublisher implements LeaderEventPublisher {