INT-4058: Add leader initiator for lock registry
JIRA: https://jira.spring.io/browse/INT-4058 If you hold the lock, you are the leader. This simple idea gets you a long way if there is no "native" leader initiator. E.g. you can use this with a RDBMS with JdbcLockRegistry. Add some docs on leader election under "endpoints" Make thread name for leader initiator unique In case there are multiple instances in the same context we would like to be able to spot them in the logs. INT-4058: Polishing * Code refactoring in the `LockRegistryLeaderInitiator`: SI use 120 line length * Add `Assert.notNull()` for required properties * Add `this.lock.unlock()` and `Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis)` into the `catch` block to let distributed elections to work. Otherwise there is a big chance that we will acquire the lock and become a leader again just after `yield()` * Change `LockContext.toString()` to use simple `String` concatenation which is optimized by compiler to the `StringBuilder`. That let to have some better micro-performance compared with with extra `Formatter` object in case of `String.format()` * Add `JdbcLockRegistryLeaderInitiatorTests` * Fix `yield()` logic based on the `Future.cancel(true)`. Since one `FutureTask.cancel(true)` makes it as `INTERRUPTED` any subsequent `yield()` does not make any effect, therefore our infinite selector loop isn't interrupted one more time. * Reschedule the `LeaderSelector` in the `yield()` after cancel(true). * Rework `catch` in the selector loop just to the `InterruptedException` and `return null;` to stop looping and let reschedule the selector. * Add one more `onRevoked` logic into the `final` of the selector loop to notify that we have lost leadership during `stop()` * Improve `JdbcLockRegistryLeaderInitiatorTests` to ensure that several `yield()` on the same initiator work well. * Add `What's New` note.
This commit is contained in:
Dave Syer
@@ -0,0 +1,423 @@
|
||||
/*
|
||||
* Copyright 2016 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.support.leader;
|
||||
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.context.ApplicationEventPublisher;
|
||||
import org.springframework.context.ApplicationEventPublisherAware;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.integration.leader.Candidate;
|
||||
import org.springframework.integration.leader.Context;
|
||||
import org.springframework.integration.leader.DefaultCandidate;
|
||||
import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
|
||||
import org.springframework.integration.leader.event.LeaderEventPublisher;
|
||||
import org.springframework.integration.leader.event.OnGrantedEvent;
|
||||
import org.springframework.integration.leader.event.OnRevokedEvent;
|
||||
import org.springframework.integration.support.locks.LockRegistry;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Component that initiates leader election based on holding a lock. If the lock has the
|
||||
* right properties (global with expiry), there will never be more than one leader, but
|
||||
* there may occasionally be no leader for short periods. If the lock has stronger
|
||||
* guarantees, and it interrupts the holder's thread when it expires or is stolen, then
|
||||
* you can adjust the parameters to reduce the leaderless period to be limited only by
|
||||
* latency to the lock provider. The election process ties up a thread perpetually while
|
||||
* we hold and try to acquire the lock, so a native leader initiator (not based on a lock)
|
||||
* is likely to be more efficient. If there is no native leader initiator available, but
|
||||
* there is a lock registry (e.g. on a shared database), this implementation is likely to
|
||||
* be useful.
|
||||
*
|
||||
* @author Dave Syer
|
||||
* @since 4.3.1
|
||||
*/
|
||||
public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware {
|
||||
|
||||
public static final long DEFAULT_HEART_BEAT_TIME = 500L;
|
||||
|
||||
public static final long DEFAULT_BUSY_WAIT_TIME = 50L;
|
||||
|
||||
private static final Log logger = LogFactory.getLog(LockRegistryLeaderInitiator.class);
|
||||
|
||||
private static int threadNameCount = 0;
|
||||
|
||||
private static final Context NULL_CONTEXT = new NullContext();
|
||||
|
||||
private final Object lifecycleMonitor = new Object();
|
||||
|
||||
/**
|
||||
* Executor service for running leadership daemon.
|
||||
*/
|
||||
private final ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
|
||||
|
||||
@Override
|
||||
public Thread newThread(Runnable r) {
|
||||
Thread thread = new Thread(r, "lock-leadership-" + (threadNameCount++));
|
||||
thread.setDaemon(true);
|
||||
return thread;
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
private final LockRegistry locks;
|
||||
|
||||
/**
|
||||
* Candidate for leader election. User injects this to receive callbacks on leadership
|
||||
* events. Alternatively applications can listen for the {@link OnGrantedEvent} and
|
||||
* {@link OnRevokedEvent}, as long as the
|
||||
* {@link #setLeaderEventPublisher(LeaderEventPublisher) leaderEventPublisher} is set.
|
||||
*/
|
||||
private final Candidate candidate;
|
||||
|
||||
|
||||
/**
|
||||
* Time in milliseconds to wait in between attempts to re-acquire the lock, once it is
|
||||
* held. The heartbeat time has to be less than the remote lock expiry period, if
|
||||
* there is one, otherwise other nodes can steal the lock while we are sleeping here.
|
||||
* If the remote lock does not expire, or if you know it interrupts the current thread
|
||||
* when it expires or is broken, then you can extend the heartbeat to Long.MAX_VALUE.
|
||||
*/
|
||||
private long heartBeatMillis = DEFAULT_HEART_BEAT_TIME;
|
||||
|
||||
/**
|
||||
* Time in milliseconds to wait in between attempts to acquire the lock, if it is not
|
||||
* held. The longer this is, the longer the system can be leaderless, if the leader
|
||||
* dies. If a leader dies without releasing its lock, the system might still have to
|
||||
* wait for the old lock to expire, but after that it should not have to wait longer
|
||||
* than the busy wait time to get a new leader. If the remote lock does not expire, or
|
||||
* if you know it interrupts the current thread when it expires or is broken, then you
|
||||
* can reduce the busy wait to zero.
|
||||
*/
|
||||
private long busyWaitMillis = DEFAULT_BUSY_WAIT_TIME;
|
||||
|
||||
private LeaderSelector leaderSelector;
|
||||
|
||||
private ApplicationEventPublisher applicationEventPublisher;
|
||||
|
||||
/**
|
||||
* Leader event publisher if set.
|
||||
*/
|
||||
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;
|
||||
|
||||
/**
|
||||
* @see SmartLifecycle which is an extension of org.springframework.context.Phased
|
||||
*/
|
||||
private volatile int phase;
|
||||
|
||||
/**
|
||||
* Flag that indicates whether the leadership election for this {@link #candidate} is
|
||||
* running.
|
||||
*/
|
||||
private volatile boolean running;
|
||||
|
||||
/**
|
||||
* Create a new leader initiator with the provided lock registry and a default
|
||||
* candidate (which just logs the leadership events).
|
||||
* @param locks lock registry
|
||||
*/
|
||||
public LockRegistryLeaderInitiator(LockRegistry locks) {
|
||||
this(locks, new DefaultCandidate());
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new leader initiator. The candidate implementation is provided by the user
|
||||
* to listen for leadership events and carry out business actions.
|
||||
*
|
||||
* @param locks lock registry
|
||||
* @param candidate leadership election candidate
|
||||
*/
|
||||
public LockRegistryLeaderInitiator(LockRegistry locks, Candidate candidate) {
|
||||
Assert.notNull(locks, "'locks' must not be null");
|
||||
Assert.notNull(candidate, "'candidate' must not be null");
|
||||
this.locks = locks;
|
||||
this.candidate = candidate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
|
||||
this.applicationEventPublisher = applicationEventPublisher;
|
||||
}
|
||||
|
||||
public void setHeartBeatMillis(long heartBeatMillis) {
|
||||
this.heartBeatMillis = heartBeatMillis;
|
||||
}
|
||||
|
||||
public void setBusyWaitMillis(long busyWaitMillis) {
|
||||
this.busyWaitMillis = busyWaitMillis;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the {@link LeaderEventPublisher}.
|
||||
* @param leaderEventPublisher the event publisher
|
||||
*/
|
||||
public void setLeaderEventPublisher(LeaderEventPublisher leaderEventPublisher) {
|
||||
this.leaderEventPublisher = leaderEventPublisher;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true if leadership election for this {@link #candidate} is running.
|
||||
*/
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return this.running;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return this.phase;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param phase the phase
|
||||
* @see SmartLifecycle
|
||||
*/
|
||||
public void setPhase(int phase) {
|
||||
this.phase = phase;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return this.autoStartup;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param autoStartup true to start automatically
|
||||
* @see SmartLifecycle
|
||||
*/
|
||||
public void setAutoStartup(boolean autoStartup) {
|
||||
this.autoStartup = autoStartup;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the context (or null if not running)
|
||||
*/
|
||||
public Context getContext() {
|
||||
if (this.leaderSelector == null) {
|
||||
return NULL_CONTEXT;
|
||||
}
|
||||
return this.leaderSelector.context;
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the registration of the {@link #candidate} for leader election.
|
||||
*/
|
||||
@Override
|
||||
public void start() {
|
||||
if (this.leaderEventPublisher == null && this.applicationEventPublisher != null) {
|
||||
this.leaderEventPublisher = new DefaultLeaderEventPublisher(this.applicationEventPublisher);
|
||||
}
|
||||
synchronized (this.lifecycleMonitor) {
|
||||
if (!this.running) {
|
||||
this.leaderSelector = new LeaderSelector(buildLeaderPath());
|
||||
this.future = this.executorService.submit(this.leaderSelector);
|
||||
this.running = true;
|
||||
logger.debug("Started LeaderInitiator");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
stop();
|
||||
this.executorService.shutdown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable runnable) {
|
||||
stop();
|
||||
runnable.run();
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the registration of the {@link #candidate} for leader election. If the
|
||||
* candidate is currently leader, its leadership will be revoked.
|
||||
*/
|
||||
@Override
|
||||
public void stop() {
|
||||
synchronized (this.lifecycleMonitor) {
|
||||
if (this.running) {
|
||||
this.running = false;
|
||||
this.future.cancel(true);
|
||||
logger.debug("Stopped LeaderInitiator");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the lock key used by leader election
|
||||
*/
|
||||
private String buildLeaderPath() {
|
||||
return this.candidate.getRole();
|
||||
}
|
||||
|
||||
protected class LeaderSelector implements Callable<Void> {
|
||||
|
||||
private final Lock lock;
|
||||
|
||||
private final String lockKey;
|
||||
|
||||
private final LockContext context = new LockContext();
|
||||
|
||||
private volatile boolean locked = false;
|
||||
|
||||
LeaderSelector(String lockKey) {
|
||||
this.lock = LockRegistryLeaderInitiator.this.locks.obtain(lockKey);
|
||||
this.lockKey = lockKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Void call() throws Exception {
|
||||
try {
|
||||
while (LockRegistryLeaderInitiator.this.running) {
|
||||
try {
|
||||
// We always try to acquire the lock, in case it expired
|
||||
boolean acquired = this.lock.tryLock(LockRegistryLeaderInitiator.this.heartBeatMillis,
|
||||
TimeUnit.MILLISECONDS);
|
||||
if (!this.locked) {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
else if (acquired) {
|
||||
// If we were able to acquire it but we were already locked we
|
||||
// should release it
|
||||
this.lock.unlock();
|
||||
// Give it a chance to expire.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.heartBeatMillis);
|
||||
}
|
||||
else {
|
||||
// Try again quickly in case the lock holder dropped it
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
}
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
if (this.locked) {
|
||||
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());
|
||||
}
|
||||
// Give it a chance to elect some other leader.
|
||||
Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis);
|
||||
Thread.currentThread().interrupt();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
finally {
|
||||
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());
|
||||
}
|
||||
}
|
||||
this.locked = false;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public boolean isLeader() {
|
||||
return this.locked;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of leadership context backed by lock registry.
|
||||
*/
|
||||
private class LockContext implements Context {
|
||||
|
||||
@Override
|
||||
public boolean isLeader() {
|
||||
return LockRegistryLeaderInitiator.this.leaderSelector.isLeader();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void yield() {
|
||||
if (LockRegistryLeaderInitiator.this.future != null) {
|
||||
LockRegistryLeaderInitiator.this.future.cancel(true);
|
||||
LockRegistryLeaderInitiator.this.future =
|
||||
LockRegistryLeaderInitiator.this.executorService
|
||||
.submit(LockRegistryLeaderInitiator.this.leaderSelector);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "LockContext{role=" + LockRegistryLeaderInitiator.this.candidate.getRole() +
|
||||
", id=" + LockRegistryLeaderInitiator.this.candidate.getId() +
|
||||
", isLeader=" + isLeader() + "}";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static final class NullContext implements Context {
|
||||
|
||||
@Override
|
||||
public boolean isLeader() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void yield() {
|
||||
// No-op
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
/*
|
||||
* Copyright 2012-2015 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.support.leader;
|
||||
|
||||
import static org.hamcrest.CoreMatchers.is;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.integration.leader.Context;
|
||||
import org.springframework.integration.leader.DefaultCandidate;
|
||||
import org.springframework.integration.leader.event.LeaderEventPublisher;
|
||||
import org.springframework.integration.support.locks.DefaultLockRegistry;
|
||||
import org.springframework.integration.support.locks.LockRegistry;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class LockRegistryLeaderInitiatorTests {
|
||||
|
||||
private CountDownLatch granted = new CountDownLatch(1);
|
||||
private CountDownLatch revoked = new CountDownLatch(1);
|
||||
private LockRegistry registry = new DefaultLockRegistry();
|
||||
private CountingPublisher publisher = new CountingPublisher(this.granted, this.revoked);
|
||||
|
||||
private LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(
|
||||
this.registry, new DefaultCandidate());
|
||||
|
||||
@Before
|
||||
public void init() {
|
||||
this.initiator.setLeaderEventPublisher(this.publisher);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void startAndStop() throws Exception {
|
||||
assertThat(this.initiator.getContext().isLeader(), is(false));
|
||||
this.initiator.start();
|
||||
assertThat(this.initiator.isRunning(), is(true));
|
||||
this.granted.await(2, TimeUnit.SECONDS);
|
||||
assertThat(this.initiator.getContext().isLeader(), is(true));
|
||||
Thread.sleep(200L);
|
||||
assertThat(this.initiator.getContext().isLeader(), is(true));
|
||||
this.initiator.stop();
|
||||
this.revoked.await(2, TimeUnit.SECONDS);
|
||||
assertThat(this.initiator.getContext().isLeader(), is(false));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void yield() throws Exception {
|
||||
assertThat(this.initiator.getContext().isLeader(), is(false));
|
||||
this.initiator.start();
|
||||
assertThat(this.initiator.isRunning(), is(true));
|
||||
this.granted.await(2, TimeUnit.SECONDS);
|
||||
assertThat(this.initiator.getContext().isLeader(), is(true));
|
||||
this.initiator.getContext().yield();
|
||||
assertThat(this.revoked.await(2, TimeUnit.SECONDS), is(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
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();
|
||||
assertThat(this.granted.await(2, TimeUnit.SECONDS), is(true));
|
||||
another.start();
|
||||
this.initiator.stop();
|
||||
assertThat(other.await(2, TimeUnit.SECONDS), is(true));
|
||||
assertThat(another.getContext().isLeader(), is(true));
|
||||
}
|
||||
|
||||
private static class CountingPublisher implements LeaderEventPublisher {
|
||||
private CountDownLatch granted;
|
||||
private CountDownLatch revoked;
|
||||
|
||||
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
|
||||
this.granted = granted;
|
||||
this.revoked = revoked;
|
||||
}
|
||||
|
||||
CountingPublisher(CountDownLatch granted) {
|
||||
this(granted, new CountDownLatch(1));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
this.revoked.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnGranted(Object source, Context context, String role) {
|
||||
this.granted.countDown();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,193 @@
|
||||
/*
|
||||
* Copyright 2016 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.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.jdbc.leader;
|
||||
|
||||
import static org.hamcrest.Matchers.is;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
|
||||
import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
|
||||
import org.springframework.integration.leader.Context;
|
||||
import org.springframework.integration.leader.DefaultCandidate;
|
||||
import org.springframework.integration.leader.event.LeaderEventPublisher;
|
||||
import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
|
||||
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
|
||||
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @since 4.3.1
|
||||
*/
|
||||
public class JdbcLockRegistryLeaderInitiatorTests {
|
||||
|
||||
public static EmbeddedDatabase dataSource;
|
||||
|
||||
@BeforeClass
|
||||
public static void init() {
|
||||
dataSource = new EmbeddedDatabaseBuilder()
|
||||
.setType(EmbeddedDatabaseType.H2)
|
||||
.addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql")
|
||||
.build();
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void destroy() {
|
||||
dataSource.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDistributedLeaderElection() throws Exception {
|
||||
CountDownLatch granted = new CountDownLatch(1);
|
||||
CountingPublisher countingPublisher = new CountingPublisher(granted);
|
||||
List<LockRegistryLeaderInitiator> initiators = new ArrayList<LockRegistryLeaderInitiator>();
|
||||
for (int i = 0; i < 2; i++) {
|
||||
DefaultLockRepository lockRepository = new DefaultLockRepository(dataSource);
|
||||
lockRepository.afterPropertiesSet();
|
||||
LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(
|
||||
new JdbcLockRegistry(lockRepository),
|
||||
new DefaultCandidate("foo", "bar"));
|
||||
initiator.setLeaderEventPublisher(countingPublisher);
|
||||
initiators.add(initiator);
|
||||
}
|
||||
|
||||
for (LockRegistryLeaderInitiator initiator : initiators) {
|
||||
initiator.start();
|
||||
}
|
||||
|
||||
assertThat(granted.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
LockRegistryLeaderInitiator initiator1 = countingPublisher.initiator;
|
||||
|
||||
LockRegistryLeaderInitiator initiator2 = null;
|
||||
|
||||
for (LockRegistryLeaderInitiator initiator : initiators) {
|
||||
if (initiator != initiator1) {
|
||||
initiator2 = initiator;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
assertNotNull(initiator2);
|
||||
|
||||
assertThat(initiator1.getContext().isLeader(), is(true));
|
||||
assertThat(initiator2.getContext().isLeader(), is(false));
|
||||
|
||||
final CountDownLatch granted1 = new CountDownLatch(1);
|
||||
final CountDownLatch granted2 = new CountDownLatch(1);
|
||||
CountDownLatch revoked1 = new CountDownLatch(1);
|
||||
CountDownLatch revoked2 = new CountDownLatch(1);
|
||||
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1) {
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
try {
|
||||
// It's difficult to see round-robin election, so block one initiator until the second is elected.
|
||||
assertThat(granted2.await(10, TimeUnit.SECONDS), is(true));
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
// No op
|
||||
}
|
||||
super.publishOnRevoked(source, context, role);
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2) {
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
try {
|
||||
// It's difficult to see round-robin election, so block one initiator until the second is elected.
|
||||
assertThat(granted1.await(10, TimeUnit.SECONDS), is(true));
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
// No op
|
||||
}
|
||||
super.publishOnRevoked(source, context, role);
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
initiator1.getContext().yield();
|
||||
|
||||
assertThat(revoked1.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
assertThat(initiator2.getContext().isLeader(), is(true));
|
||||
assertThat(initiator1.getContext().isLeader(), is(false));
|
||||
|
||||
initiator2.getContext().yield();
|
||||
|
||||
assertThat(revoked2.await(10, TimeUnit.SECONDS), is(true));
|
||||
|
||||
assertThat(initiator1.getContext().isLeader(), is(true));
|
||||
assertThat(initiator2.getContext().isLeader(), is(false));
|
||||
|
||||
initiator2.stop();
|
||||
|
||||
CountDownLatch revoked11 = new CountDownLatch(1);
|
||||
initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11));
|
||||
|
||||
initiator1.getContext().yield();
|
||||
|
||||
assertThat(revoked11.await(10, TimeUnit.SECONDS), is(true));
|
||||
assertThat(initiator1.getContext().isLeader(), is(false));
|
||||
|
||||
initiator1.stop();
|
||||
}
|
||||
|
||||
private static class CountingPublisher implements LeaderEventPublisher {
|
||||
|
||||
private CountDownLatch granted;
|
||||
|
||||
private CountDownLatch revoked;
|
||||
|
||||
private volatile LockRegistryLeaderInitiator initiator;
|
||||
|
||||
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
|
||||
this.granted = granted;
|
||||
this.revoked = revoked;
|
||||
}
|
||||
|
||||
CountingPublisher(CountDownLatch granted) {
|
||||
this(granted, new CountDownLatch(1));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnRevoked(Object source, Context context, String role) {
|
||||
this.revoked.countDown();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void publishOnGranted(Object source, Context context, String role) {
|
||||
this.initiator = (LockRegistryLeaderInitiator) source;
|
||||
this.granted.countDown();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -37,7 +37,7 @@ import org.springframework.util.StringUtils;
|
||||
* @author Patrick Peralta
|
||||
* @author Janne Valkealahti
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 4.2
|
||||
*/
|
||||
public class LeaderInitiator implements SmartLifecycle {
|
||||
|
||||
@@ -68,7 +68,7 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
private volatile boolean autoStartup = true;
|
||||
|
||||
/**
|
||||
* @See SmartLifecycle which is an extension of org.springframework.context.Phased
|
||||
* @see SmartLifecycle which is an extension of org.springframework.context.Phased
|
||||
*/
|
||||
private volatile int phase;
|
||||
|
||||
@@ -214,7 +214,7 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
/**
|
||||
* Implementation of Curator leadership election listener.
|
||||
*/
|
||||
class LeaderListener extends LeaderSelectorListenerAdapter {
|
||||
protected class LeaderListener extends LeaderSelectorListenerAdapter {
|
||||
|
||||
@Override
|
||||
public void takeLeadership(CuratorFramework framework) throws Exception {
|
||||
@@ -223,7 +223,8 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
try {
|
||||
LeaderInitiator.this.candidate.onGranted(context);
|
||||
if (LeaderInitiator.this.leaderEventPublisher != null) {
|
||||
LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, context, LeaderInitiator.this.candidate.getRole());
|
||||
LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, context,
|
||||
LeaderInitiator.this.candidate.getRole());
|
||||
}
|
||||
|
||||
// when this method exits, the leadership will be revoked;
|
||||
@@ -239,7 +240,8 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
finally {
|
||||
LeaderInitiator.this.candidate.onRevoked(context);
|
||||
if (LeaderInitiator.this.leaderEventPublisher != null) {
|
||||
LeaderInitiator.this.leaderEventPublisher.publishOnRevoked(LeaderInitiator.this, context, LeaderInitiator.this.candidate.getRole());
|
||||
LeaderInitiator.this.leaderEventPublisher.publishOnRevoked(LeaderInitiator.this, context,
|
||||
LeaderInitiator.this.candidate.getRole());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -248,7 +250,7 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
/**
|
||||
* Implementation of leadership context backed by Curator.
|
||||
*/
|
||||
class CuratorContext implements Context {
|
||||
private class CuratorContext implements Context {
|
||||
|
||||
@Override
|
||||
public boolean isLeader() {
|
||||
@@ -262,8 +264,9 @@ public class LeaderInitiator implements SmartLifecycle {
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.format("CuratorContext{role=%s, id=%s, isLeader=%s}",
|
||||
LeaderInitiator.this.candidate.getRole(), LeaderInitiator.this.candidate.getId(), isLeader());
|
||||
return "LockContext{role=" + LeaderInitiator.this.candidate.getRole() +
|
||||
", id=" + LeaderInitiator.this.candidate.getId() +
|
||||
", isLeader=" + isLeader() + "}";
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -630,4 +630,26 @@ The `SmartLifecycleRoleController` implements `ApplicationListener<AbstractLeade
|
||||
start/stop its configured `SmartLifecycle` objects when leadership is granted/revoked (when some bean publishes
|
||||
`OnGrantedEvent` or `OnRevokedEvent` respectively).
|
||||
|
||||
See <<zk-leadership>> for more information about leadership election and events.
|
||||
[[leadership-event-handling]]
|
||||
=== Leadership Event Handling
|
||||
|
||||
Groups of endpoints can be started/stopped based on leadership being granted or revoked respectively.
|
||||
This is useful in clustered scenarios where shared resources must only be consumed by a single instance.
|
||||
An example of this is a file inbound channel adapter that is polling a shared directory.
|
||||
(See <<file-reading>>).
|
||||
|
||||
To participate in a leader election and be notified when elected leader or when leadership is revoked, an application creates a component in the application context called a "leader initiator". Normally a leader initiator is a `SmartLifecycle` so it starts up (optionally) automatically when the context starts, and then publishes notifications when leadership changes. By convention the user provides a `Candidate` that receives the callbacks and also can revoke the leadership through a `Context` object provided by the framework. User code can also listen for `AbstractLeaderEvents`, and respond accordingly, for instance using a `SmartLifecycleRoleController`.
|
||||
|
||||
There is a basic implementation of a leader initiator based on the `LockRegistry` abstraction. To use it you just need to create an instance as a bean, for example:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
|
||||
return new LockRegistryLeaderInitiator(locks);
|
||||
}
|
||||
----
|
||||
|
||||
If the lock registry is implemented correctly, there will only ever be at most one leader. If the lock registry also provides locks which throw exceptions (ideally `InterruptedException`) when they expire or are broken, then the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation. By default there is a `busyWaitMillis` property that adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect and you only know they expired by trying to obtain one again.
|
||||
|
||||
See <<zk-leadership>> for more information about leadership election and events using Zookeeper.
|
||||
|
||||
@@ -48,6 +48,11 @@ See <<integration-graph>> for more information.
|
||||
A new `JdbcLockRegistry` is provided for distributed locks shared through the data base table.
|
||||
See <<jdbc-lock-registry>> for more information.
|
||||
|
||||
==== Leader Initiator for Lock Registry
|
||||
|
||||
A new `LeaderInitiator` implementation is provided based on the `LockRegistry` strategy.
|
||||
See <<leadership-event-handling>> for more information.
|
||||
|
||||
[[x4.3-general]]
|
||||
=== General Changes
|
||||
|
||||
|
||||
@@ -64,10 +64,7 @@ to time, to remove old unused locks from memory.
|
||||
[[zk-leadership]]
|
||||
=== Zookeeper Leadership Event Handling
|
||||
|
||||
Groups of endpoints can be started/stopped based on leadership being granted or revoked respectively.
|
||||
This is useful in clustered scenarios where shared resources must only be consumed by a single instance.
|
||||
An example of this is a file inbound channel adapter that is polling a shared directory.
|
||||
(See <<file-reading>>).
|
||||
To configure an application for leader election using Zookeeper in XML:
|
||||
|
||||
[source, xml]
|
||||
----
|
||||
@@ -81,6 +78,8 @@ When leadership is revoked, an `OnRevokedEvent` will be published for the role `
|
||||
will be stopped.
|
||||
See <<endpoint-roles>> for more information.
|
||||
|
||||
In Java configuration you can create an instance of the leader initiator like this:
|
||||
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
|
||||
Reference in New Issue
Block a user