Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2354 Injection Executor to LockLeaderInitiator

Browse Source 
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
This commit is contained in:
Kiel Boatman
2018-02-10 10:04:44 +00:00
committed by Artem Bilan
parent d361eefa0c
commit 78b80ef79f
2 changed files with 84 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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

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

@@ -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 {