INT-4220: Leaders: warn event publishing errors

JIRA: https://jira.spring.io/browse/INT-4220

Currently when an error is thrown from the event publishing the role granting is broken and we just go to the role revoking.

* Since it's just an event publishing it shouldn't effect the original leader election.
* `try...catch` event publishing in the `LeaderInitiator` and `logger.warn` an `Exception`
* Make `leader/Context` as `@FunctionalInterface` for simple Lambda use-case like `NULL_CONTEXT` - `() -> false`
* Remove all the internal `NullContext` implementations in favor of above mention Lambda
* In the `LockRegistryLeaderInitiator` use `CustomizableThreadFactory` instead of custom `ThreadFactory` for prefixing
* Add `zookeeper/leader/LeaderInitiator#getContext()` for external usage and consistency with other similar components
* Fix `CuratorContext.toString()` typo

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.
+ * <p>
+ * 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
+	}
+
 }

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

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;

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

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() + "}";
 		}

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 {