From 78b80ef79fed817143833a5f0d08b2753966a300 Mon Sep 17 00:00:00 2001 From: Kiel Boatman Date: Sat, 10 Feb 2018 10:04:44 +0000 Subject: [PATCH] GH-2354 Injection Executor to LockLeaderInitiator Fixes https://github.com/spring-projects/spring-integration/issues/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 --- .../leader/LockRegistryLeaderInitiator.java | 59 +++++++++++------ .../LockRegistryLeaderInitiatorTests.java | 63 +++++++++++++------ 2 files changed, 84 insertions(+), 38 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java index c1ffd0b18d..9295c7854d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java @@ -57,6 +57,7 @@ import org.springframework.util.Assert; * @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,21 +130,15 @@ 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 @@ -146,6 +146,12 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe */ 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 class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe 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 class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } /** - * 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 class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } @Override - public void destroy() throws Exception { + public void destroy() { stop(); - this.executorService.shutdown(); + if (!this.executorServiceExplicitlySet) { + this.executorService.shutdown(); + } } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java index f4957407b5..6c9257c262 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java +++ b/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.mock; 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.DefaultLeaderEventPublisher; 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 class LockRegistryLeaderInitiatorTests { 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 class LockRegistryLeaderInitiatorTests { 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 class LockRegistryLeaderInitiatorTests { }); - 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 class LockRegistryLeaderInitiatorTests { // 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 class LockRegistryLeaderInitiatorTests { // 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 class LockRegistryLeaderInitiatorTests { 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 class LockRegistryLeaderInitiatorTests { 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 {