diff --git a/spring-integration-core/src/main/java/org/springframework/integration/leader/Context.java b/spring-integration-core/src/main/java/org/springframework/integration/leader/Context.java index 5bf7932ee5..dbc6990eef 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/leader/Context.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/leader/Context.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -20,17 +20,21 @@ package org.springframework.integration.leader; * Interface that defines the context for candidate leadership. * Instances of this object are passed to {@link Candidate candidates} * upon granting and revoking of leadership. + *

+ * The {@link Context} is {@link FunctionalInterface} + * with no-op implementation for the {@link #yield()}. * * @author Patrick Peralta * @author Janne Valkealahti + * @author Artem Bilan * */ +@FunctionalInterface public interface Context { /** * Checks if the {@link Candidate} this context was * passed to is the leader. - * * @return true if the {@link Candidate} this context was * passed to is the leader */ @@ -41,5 +45,8 @@ public interface Context { * to relinquish leadership. This method has no effect * if the candidate is not currently the leader. */ - void yield(); + default void yield() { + // no-op + } + } 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 7502bcd6f7..806a57f0fd 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -20,7 +20,6 @@ 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; @@ -39,6 +38,7 @@ 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.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; /** @@ -55,6 +55,7 @@ import org.springframework.util.Assert; * * @author Dave Syer * @author Artem Bilan + * * @since 4.3.1 */ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware { @@ -65,25 +66,15 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe private static final Log logger = LogFactory.getLog(LockRegistryLeaderInitiator.class); - private static int threadNameCount = 0; - - private static final Context NULL_CONTEXT = new NullContext(); + private static final Context NULL_CONTEXT = () -> false; 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; - } - - }); + 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 @@ -163,7 +154,6 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe /** * 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 */ @@ -325,8 +315,13 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe this.locked = true; LockRegistryLeaderInitiator.this.candidate.onGranted(this.context); if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) { - LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted( - LockRegistryLeaderInitiator.this, this.context, this.lockKey); + try { + LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnGranted( + LockRegistryLeaderInitiator.this, this.context, this.lockKey); + } + catch (Exception e) { + logger.warn("Error publishing OnGranted event.", e); + } } } } @@ -349,9 +344,14 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe // 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()); + try { + LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LockRegistryLeaderInitiator.this, this.context, + LockRegistryLeaderInitiator.this.candidate.getRole()); + } + catch (Exception e1) { + logger.warn("Error publishing OnRevoked event.", e); + } } // Give it a chance to elect some other leader. Thread.sleep(LockRegistryLeaderInitiator.this.busyWaitMillis); @@ -367,9 +367,14 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe // 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()); + try { + LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnRevoked( + LockRegistryLeaderInitiator.this, this.context, + LockRegistryLeaderInitiator.this.candidate.getRole()); + } + catch (Exception e) { + logger.warn("Error publishing OnRevoked event.", e); + } } } this.locked = false; @@ -418,22 +423,4 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe } - private static final class NullContext implements Context { - - NullContext() { - super(); - } - - @Override - public boolean isLeader() { - return false; - } - - @Override - public void yield() { - // No-op - } - - } - } 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 f971252552..7e93bed33d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2012-2016 the original author or authors. + * Copyright 2012-2017 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. @@ -18,21 +18,20 @@ package org.springframework.integration.support.leader; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.apache.log4j.Level; import org.junit.Before; -import org.junit.Rule; import org.junit.Test; 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.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; -import org.springframework.integration.test.rule.Log4jLevelAdjuster; /** * @author Dave Syer @@ -42,9 +41,6 @@ import org.springframework.integration.test.rule.Log4jLevelAdjuster; */ public class LockRegistryLeaderInitiatorTests { - @Rule - public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, "org.springframework.integration"); - private CountDownLatch granted; private CountDownLatch revoked; @@ -84,6 +80,7 @@ public class LockRegistryLeaderInitiatorTests { assertThat(this.initiator.getContext().isLeader(), is(true)); this.initiator.getContext().yield(); assertThat(this.revoked.await(10, TimeUnit.SECONDS), is(true)); + this.initiator.stop(); } @Test @@ -100,6 +97,34 @@ public class LockRegistryLeaderInitiatorTests { assertThat(another.getContext().isLeader(), is(true)); } + @Test + public void testExceptionFromEvent() throws Exception { + CountDownLatch onGranted = new CountDownLatch(1); + + LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate()); + + initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() { + + @Override + public void publishOnGranted(Object source, Context context, String role) { + try { + throw new RuntimeException("intentional"); + } + finally { + onGranted.countDown(); + } + } + + }); + + initiator.start(); + + assertTrue(onGranted.await(10, TimeUnit.SECONDS)); + assertTrue(initiator.getContext().isLeader()); + + initiator.stop(); + } + private static class CountingPublisher implements LeaderEventPublisher { private final CountDownLatch granted; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java index 450b490299..454a0d3622 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -25,10 +25,8 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.apache.log4j.Level; import org.junit.AfterClass; import org.junit.BeforeClass; -import org.junit.Rule; import org.junit.Test; import org.springframework.integration.jdbc.lock.DefaultLockRepository; @@ -37,7 +35,6 @@ 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.integration.test.rule.Log4jLevelAdjuster; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; @@ -48,9 +45,6 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; */ public class JdbcLockRegistryLeaderInitiatorTests { - @Rule - public Log4jLevelAdjuster logAdjuster = new Log4jLevelAdjuster(Level.DEBUG, "org.springframework.integration"); - public static EmbeddedDatabase dataSource; @BeforeClass diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java index 8a4854f3a5..56e93bc3a4 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/leader/LeaderInitiator.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -37,6 +37,8 @@ import org.springframework.util.StringUtils; * @author Patrick Peralta * @author Janne Valkealahti * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 */ public class LeaderInitiator implements SmartLifecycle { @@ -45,6 +47,10 @@ public class LeaderInitiator implements SmartLifecycle { private static final String DEFAULT_NAMESPACE = "/spring-integration/leader/"; + private static final Context NULL_CONTEXT = () -> false; + + private final CuratorContext context = new CuratorContext(); + /** * Curator client. */ @@ -196,6 +202,18 @@ public class LeaderInitiator implements SmartLifecycle { this.leaderEventPublisher = leaderEventPublisher; } + /** + * The context of the initiator or null if not running. + * @return the context (or null if not running) + * @since 5.0 + */ + public Context getContext() { + if (this.leaderSelector == null) { + return NULL_CONTEXT; + } + return this.context; + } + /** * @return the ZooKeeper path used for leadership election by Curator */ @@ -218,13 +236,16 @@ public class LeaderInitiator implements SmartLifecycle { @Override public void takeLeadership(CuratorFramework framework) throws Exception { - CuratorContext context = new CuratorContext(); - try { - LeaderInitiator.this.candidate.onGranted(context); + LeaderInitiator.this.candidate.onGranted(LeaderInitiator.this.context); if (LeaderInitiator.this.leaderEventPublisher != null) { - LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, context, - LeaderInitiator.this.candidate.getRole()); + try { + LeaderInitiator.this.leaderEventPublisher.publishOnGranted(LeaderInitiator.this, + LeaderInitiator.this.context, LeaderInitiator.this.candidate.getRole()); + } + catch (Exception e) { + logger.warn("Error publishing OnGranted event.", e); + } } // when this method exits, the leadership will be revoked; @@ -238,10 +259,15 @@ public class LeaderInitiator implements SmartLifecycle { // reset the interrupt flag as the interrupt is handled. } finally { - LeaderInitiator.this.candidate.onRevoked(context); + LeaderInitiator.this.candidate.onRevoked(LeaderInitiator.this.context); if (LeaderInitiator.this.leaderEventPublisher != null) { - LeaderInitiator.this.leaderEventPublisher.publishOnRevoked(LeaderInitiator.this, context, - LeaderInitiator.this.candidate.getRole()); + try { + LeaderInitiator.this.leaderEventPublisher.publishOnRevoked(LeaderInitiator.this, + LeaderInitiator.this.context, LeaderInitiator.this.candidate.getRole()); + } + catch (Exception e) { + logger.warn("Error publishing OnRevoked event.", e); + } } } } @@ -268,7 +294,7 @@ public class LeaderInitiator implements SmartLifecycle { @Override public String toString() { - return "LockContext{role=" + LeaderInitiator.this.candidate.getRole() + + return "CuratorContext{role=" + LeaderInitiator.this.candidate.getRole() + ", id=" + LeaderInitiator.this.candidate.getId() + ", isLeader=" + isLeader() + "}"; } diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBeanTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBeanTests.java index 0cf45a978b..6e8e440729 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBeanTests.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBeanTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -35,7 +35,10 @@ 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.DefaultLeaderEventPublisher; import org.springframework.integration.leader.event.OnGrantedEvent; import org.springframework.integration.leader.event.OnRevokedEvent; import org.springframework.integration.zookeeper.ZookeeperTestSupport; @@ -84,6 +87,34 @@ public class LeaderInitiatorFactoryBeanTests extends ZookeeperTestSupport { assertThat(this.config.events.get(1), instanceOf(OnRevokedEvent.class)); } + @Test + public void testExceptionFromEvent() throws Exception { + CountDownLatch onGranted = new CountDownLatch(1); + + LeaderInitiator initiator = new LeaderInitiator(client, new DefaultCandidate()); + + initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() { + + @Override + public void publishOnGranted(Object source, Context context, String role) { + try { + throw new RuntimeException("intentional"); + } + finally { + onGranted.countDown(); + } + } + + }); + + initiator.start(); + + assertTrue(onGranted.await(10, TimeUnit.SECONDS)); + assertTrue(initiator.getContext().isLeader()); + + initiator.stop(); + } + @Configuration public static class Config {