diff --git a/spring-integration-hazelcast/README.md b/spring-integration-hazelcast/README.md index c01d6cc..303f80b 100644 --- a/spring-integration-hazelcast/README.md +++ b/spring-integration-hazelcast/README.md @@ -458,3 +458,25 @@ public HazelcastCacheWritingMessageHandler hazelcastCacheWritingMessageHandler() } ``` + +## HAZELCAST LEADER ELECTION + +If you need to elect a leader (e.g. for highly available message consumer where only one node should receive messages) +you just need to create a `LeaderInitiator`. Example: + +```java +@Bean +public HazelcastInstance hazelcastInstance() { + return Hazelcast.newHazelcastInstance(); +} + +@Bean +public LeaderInitiator initiator() { + LeaderInitiator initiator = new LeaderInitiator(hazelcastInstance()); + return initiator; +} +``` + +Then when a node is elected leader it will send `OnGrantedEvent` to all application listeners. See +the [Spring Integration User Guide](http://docs.spring.io/spring-integration/reference/htmlsingle/#endpoint-roles) +for more information on how to use those events to control messaging endpoints. diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java new file mode 100644 index 0000000..d01c2a5 --- /dev/null +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java @@ -0,0 +1,331 @@ +/* + * Copyright 2014-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.hazelcast.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 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.util.Assert; + +import com.hazelcast.core.HazelcastInstance; +import com.hazelcast.core.ILock; + +/** + * Bootstrap leadership {@link org.springframework.integration.leader.Candidate candidates} + * with Hazelcast. Upon construction, {@link #start} must be invoked to + * register the candidate for leadership election. + * + * @author Patrick Peralta + * @author Gary Russell + * @author Dave Syer + * @author Artem Bilan + */ +public class LeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware { + + private static int threadNameCount = 0; + + private static final Context NULL_CONTEXT = new NullContext(); + + /** + * Hazelcast client. + */ + private final HazelcastInstance client; + + /** + * Candidate for leader election. + */ + private final Candidate candidate; + + /** + * 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, "Hazelcast-leadership-" + (threadNameCount++)); + thread.setDaemon(true); + return thread; + } + + }); + + /** + * Future returned by submitting an {@link LeaderSelector} to {@link #executorService}. + * This is used to cancel leadership. + */ + private volatile Future future; + + /** + * Hazelcast distributed lock. + */ + private volatile ILock lock; + + private LeaderSelector leaderSelector; + + /** + * Leader event publisher. + */ + private volatile LeaderEventPublisher leaderEventPublisher = new DefaultLeaderEventPublisher(); + + private boolean customPublisher = false; + + /** + * @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; + + /** + * Construct a {@link LeaderInitiator} with a default candidate. + * + * @param client Hazelcast client + */ + public LeaderInitiator(HazelcastInstance client) { + this(client, new DefaultCandidate()); + } + + /** + * Construct a {@link LeaderInitiator}. + * + * @param client Hazelcast client + * @param candidate leadership election candidate + */ + public LeaderInitiator(HazelcastInstance client, Candidate candidate) { + Assert.notNull(client, "'client' must not be null"); + Assert.notNull(candidate, "'candidate' must not be null"); + this.client = client; + this.candidate = candidate; + } + + /** + * Sets the {@link LeaderEventPublisher}. + * @param leaderEventPublisher the event publisher + */ + public void setLeaderEventPublisher(LeaderEventPublisher leaderEventPublisher) { + Assert.notNull(leaderEventPublisher); + this.leaderEventPublisher = leaderEventPublisher; + this.customPublisher = true; + } + + /** + * @return the context (or null if not running) + */ + public Context getContext() { + if (this.leaderSelector == null) { + return NULL_CONTEXT; + } + return this.leaderSelector.context; + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + if (!this.customPublisher) { + this.leaderEventPublisher = new DefaultLeaderEventPublisher(applicationEventPublisher); + } + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public int getPhase() { + return this.phase; + } + + /** + * Start the registration of the {@link #candidate} for leader election. + */ + @Override + public synchronized void start() { + if (!this.running) { + this.lock = this.client.getLock(this.candidate.getRole()); + this.running = true; + this.leaderSelector = new LeaderSelector(); + this.future = this.executorService.submit(this.leaderSelector); + } + } + + @Override + public void stop(Runnable callback) { + stop(); + if (callback != null) { + callback.run(); + } + } + + /** + * Stop the registration of the {@link #candidate} for leader election. + * If the candidate is currently leader, its leadership will be revoked. + */ + @Override + public synchronized void stop() { + if (this.running) { + this.running = false; + this.future.cancel(true); + } + } + + /** + * @return true if leadership election for this {@link #candidate} is running + */ + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public void destroy() throws Exception { + stop(); + this.executorService.shutdown(); + } + + /** + * Callable that manages the acquisition of Hazelcast locks + * for leadership election. + */ + protected class LeaderSelector implements Callable { + + protected final HazelcastContext context = new HazelcastContext(); + + protected final String role = LeaderInitiator.this.candidate.getRole(); + + private volatile boolean locked = false; + + @Override + public Void call() throws Exception { + try { + while (LeaderInitiator.this.running) { + try { + this.locked = LeaderInitiator.this.lock.tryLock(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + if (this.locked) { + LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, + this.context, this.role); + LeaderInitiator.this.candidate.onGranted(context); + Thread.sleep(Long.MAX_VALUE); + } + } + catch (InterruptedException e) { + if (this.locked) { + LeaderInitiator.this.lock.unlock(); + this.locked = false; + // The lock was broken and we are no longer leader + LeaderInitiator.this.candidate.onRevoked(this.context); + if (LeaderInitiator.this.leaderEventPublisher != null) { + LeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LeaderInitiator.this, this.context, + LeaderInitiator.this.candidate.getRole()); + } + Thread.currentThread().interrupt(); + return null; + } + } + } + } + finally { + if (this.locked) { + LeaderInitiator.this.lock.unlock(); + this.locked = false; + // We are stopping, therefore not leading any more + LeaderInitiator.this.candidate.onRevoked(this.context); + if (LeaderInitiator.this.leaderEventPublisher != null) { + LeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LeaderInitiator.this, this.context, this.role); + } + } + } + return null; + } + + } + + /** + * Implementation of leadership context backed by Hazelcast. + */ + protected class HazelcastContext implements Context { + + @Override + public boolean isLeader() { + return LeaderInitiator.this.leaderSelector.locked; + } + + @Override + public void yield() { + if (LeaderInitiator.this.future != null) { + LeaderInitiator.this.future.cancel(true); + LeaderInitiator.this.future = + LeaderInitiator.this.executorService.submit(LeaderInitiator.this.leaderSelector); + } + } + + @Override + public String toString() { + return "HazelcastContext{role=" + LeaderInitiator.this.candidate.getRole() + + ", id=" + LeaderInitiator.this.candidate.getId() + + ", isLeader=" + isLeader() + "}"; + } + + } + + private static final class NullContext implements Context { + + @Override + public boolean isLeader() { + return false; + } + + @Override + public void yield() { + // No-op + } + + } + +} diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java new file mode 100644 index 0000000..d58c33c --- /dev/null +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java @@ -0,0 +1,253 @@ +/* + * Copyright 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.hazelcast.leader; + +import static org.hamcrest.CoreMatchers.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.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.leader.Context; +import org.springframework.integration.leader.DefaultCandidate; +import org.springframework.integration.leader.event.AbstractLeaderEvent; +import org.springframework.integration.leader.event.LeaderEventPublisher; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.hazelcast.core.Hazelcast; +import com.hazelcast.core.HazelcastInstance; + +/** + * Tests for hazelcast leader election. + * + * @author Janne Valkealahti + * @author Patrick Peralta + * @author Dave Syer + * @author Artem Bilan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +@DirtiesContext +public class LeaderInitiatorTests { + + @Autowired + private HazelcastInstance hazelcastInstance; + + @Autowired + private TestCandidate candidate; + + @Autowired + private TestEventListener listener; + + @Autowired + private LeaderInitiator initiator; + + @Test + public void testLeaderElections() throws Exception { + assertThat(this.candidate.onGrantedLatch.await(5, TimeUnit.SECONDS), is(true)); + assertThat(this.listener.onEventLatch.await(5, TimeUnit.SECONDS), is(true)); + assertThat(this.listener.events.size(), is(1)); + + this.initiator.destroy(); + + CountDownLatch granted = new CountDownLatch(1); + CountingPublisher countingPublisher = new CountingPublisher(granted); + List initiators = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + LeaderInitiator initiator = new LeaderInitiator(this.hazelcastInstance, new DefaultCandidate()); + initiator.setLeaderEventPublisher(countingPublisher); + initiators.add(initiator); + } + + for (LeaderInitiator initiator : initiators) { + initiator.start(); + } + + assertThat(granted.await(10, TimeUnit.SECONDS), is(true)); + + LeaderInitiator initiator1 = countingPublisher.initiator; + + LeaderInitiator initiator2 = null; + + for (LeaderInitiator 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.destroy(); + + 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.destroy(); + } + + @Configuration + public static class TestConfig { + + @Bean + public TestCandidate candidate() { + return new TestCandidate(); + } + + @Bean + public HazelcastInstance hazelcastInstance() { + return Hazelcast.newHazelcastInstance(); + } + + @Bean + public LeaderInitiator initiator() { + return new LeaderInitiator(hazelcastInstance(), candidate()); + } + + @Bean + public TestEventListener testEventListener() { + return new TestEventListener(); + } + + } + + static class TestCandidate extends DefaultCandidate { + + CountDownLatch onGrantedLatch = new CountDownLatch(1); + + @Override + public void onGranted(Context ctx) { + this.onGrantedLatch.countDown(); + super.onGranted(ctx); + } + + } + + static class TestEventListener implements ApplicationListener { + + CountDownLatch onEventLatch = new CountDownLatch(1); + + ArrayList events = new ArrayList(); + + @Override + public void onApplicationEvent(AbstractLeaderEvent event) { + this.events.add(event); + this.onEventLatch.countDown(); + } + + } + + private static class CountingPublisher implements LeaderEventPublisher { + + private CountDownLatch granted; + + private CountDownLatch revoked; + + private volatile LeaderInitiator 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 = (LeaderInitiator) source; + this.granted.countDown(); + } + + } + +}