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 index 97dee71..dcf08bc 100644 --- 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 @@ -23,6 +23,9 @@ import java.util.concurrent.Future; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +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; @@ -49,6 +52,8 @@ import com.hazelcast.core.ILock; */ public class LeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware { + private static final Log logger = LogFactory.getLog(LeaderInitiator.class); + private static int threadNameCount = 0; private static final Context NULL_CONTEXT = new NullContext(); @@ -236,8 +241,13 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat 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); + try { + LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, + this.context, this.role); + } + catch (Exception e) { + logger.warn("Error publishing OnGranted event.", e); + } LeaderInitiator.this.candidate.onGranted(this.context); Thread.sleep(Long.MAX_VALUE); } @@ -249,9 +259,14 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat // 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()); + try { + LeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LeaderInitiator.this, this.context, + LeaderInitiator.this.candidate.getRole()); + } + catch (Exception ex) { + logger.warn("Error publishing OnRevoked event.", ex); + } } Thread.currentThread().interrupt(); return null; @@ -266,8 +281,13 @@ public class LeaderInitiator implements SmartLifecycle, DisposableBean, Applicat // 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); + try { + LeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LeaderInitiator.this, this.context, this.role); + } + catch (Exception e) { + logger.warn("Error publishing OnRevoked event.", e); + } } } } 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 index 83afdc4..0b61013 100644 --- 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 @@ -20,6 +20,9 @@ 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 static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.spy; import java.util.ArrayList; import java.util.List; @@ -36,6 +39,7 @@ 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.DefaultLeaderEventPublisher; import org.springframework.integration.leader.event.LeaderEventPublisher; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -168,8 +172,43 @@ public class LeaderInitiatorTests { assertThat(revoked11.await(10, TimeUnit.SECONDS), is(true)); initiator1.destroy(); + + + + CountDownLatch onGranted = new CountDownLatch(1); + + DefaultCandidate candidate = spy(new DefaultCandidate()); + willAnswer(invocation -> { + try { + return invocation.callRealMethod(); + } + finally { + onGranted.countDown(); + } + }) + .given(candidate).onGranted(any(Context.class)); + + LeaderInitiator initiator = new LeaderInitiator(this.hazelcastInstance, candidate); + + initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() { + + @Override + public void publishOnGranted(Object source, Context context, String role) { + throw new RuntimeException("intentional"); + } + + }); + + initiator.start(); + + assertThat(onGranted.await(5, TimeUnit.SECONDS), is(true)); + + assertThat(initiator.getContext().isLeader(), is(true)); + + initiator.destroy(); } + @Configuration public static class TestConfig {