INT-3747: Zookeeper Documentation

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

INT-3747: Polishing; PR Comments; Factory Beans

Fix JavaDocs and simple polishing

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBean.java

@@ -0,0 +1,152 @@
+/*
+ * 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.zookeeper.config;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.utils.CloseableUtils;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.util.Assert;
+
+/**
+ * A spring-friendly way to build a {@link CuratorFramework} and implementing {@link SmartLifecycle}.
+ *
+ * @author Gary Russell
+ *
+ */
+public class CuratorFrameworkFactoryBean implements FactoryBean<CuratorFramework>, SmartLifecycle {
+
+	private final Object lifecycleLock = new Object();
+
+	private final CuratorFramework client;
+
+	/**
+	 * @see SmartLifecycle
+	 */
+	private volatile boolean autoStartup = true;
+
+	/**
+	 * @see SmartLifecycle
+	 */
+	private volatile boolean running;
+
+	/**
+	 * @see SmartLifecycle
+	 */
+	private volatile int phase;
+
+
+	/**
+	 * Construct an instance using the supplied connection string and using a default
+	 * retry policy {@code new ExponentialBackoffRetry(1000, 3)}.
+	 * @param connectionString list of servers to connect to
+	 */
+	public CuratorFrameworkFactoryBean(String connectionString) {
+		this(connectionString, new ExponentialBackoffRetry(1000, 3));
+	}
+
+	/**
+	 * Construct an instance using the supplied connection string and retry policy.
+	 * @param connectionString list of servers to connect to
+	 * @param retryPolicy the retry policy
+	 */
+	public CuratorFrameworkFactoryBean(String connectionString, RetryPolicy retryPolicy) {
+		Assert.notNull(connectionString, "'connectionString' cannot be null");
+		Assert.notNull(retryPolicy, "'retryPolicy' cannot be null");
+		this.client = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	/**
+	 * @param phase the phase
+	 * @see SmartLifecycle
+	 */
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	/**
+	 * @param autoStartup true to automatically start
+	 * @see SmartLifecycle
+	 */
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public void start() {
+		synchronized (this.lifecycleLock) {
+			if (!this.running) {
+				if (this.client != null) {
+					this.client.start();
+				}
+				this.running = true;
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		synchronized (this.lifecycleLock) {
+			if (this.running) {
+				if (this.client.getState().equals(CuratorFrameworkState.STARTED)) {
+					CloseableUtils.closeQuietly(this.client);
+				}
+			}
+		}
+	}
+
+	@Override
+	public void stop(Runnable runnable) {
+		stop();
+		runnable.run();
+	}
+
+	@Override
+	public CuratorFramework getObject() throws Exception {
+		return this.client;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return CuratorFramework.class;
+	}
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+}

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBean.java

@@ -0,0 +1,147 @@
+/*
+ * 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.zookeeper.config;
+
+import java.util.UUID;
+
+import org.apache.curator.framework.CuratorFramework;
+
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.InitializingBean;
+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.DefaultCandidate;
+import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
+import org.springframework.integration.zookeeper.leader.LeaderInitiator;
+
+/**
+ * Creates a {@link LeaderInitiator}.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class LeaderInitiatorFactoryBean
+		implements FactoryBean<LeaderInitiator>, SmartLifecycle, InitializingBean, ApplicationEventPublisherAware {
+
+	private final CuratorFramework client;
+
+	private final Candidate candidate;
+
+	private final String path;
+
+	private LeaderInitiator leaderInitiator;
+
+	private boolean autoStartup = true;
+
+	private int phase;
+
+	private ApplicationEventPublisher applicationEventPublisher;
+
+	/**
+	 * Construct the instance.
+	 * @param client the {@link CuratorFramework}.
+	 * @param path the path in zookeeper.
+	 * @param role the role of the leader.
+	 */
+	public LeaderInitiatorFactoryBean(CuratorFramework client, String path, String role) {
+		this.client = client;
+		this.candidate = new DefaultCandidate(UUID.randomUUID().toString(), role);
+		this.path = path;
+	}
+
+	@Override
+	public void start() {
+		if (this.leaderInitiator != null) {
+			this.leaderInitiator.start();
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.leaderInitiator != null) {
+			this.leaderInitiator.stop();
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.leaderInitiator != null && this.leaderInitiator.isRunning();
+	}
+
+	@Override
+	public int getPhase() {
+		if (this.leaderInitiator != null) {
+			return this.leaderInitiator.getPhase();
+		}
+		return 0;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.leaderInitiator != null && this.leaderInitiator.isAutoStartup();
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		if (this.leaderInitiator == null) {
+			this.leaderInitiator = new LeaderInitiator(this.client, this.candidate, this.path);
+			this.leaderInitiator.setPhase(this.phase);
+			this.leaderInitiator.setAutoStartup(this.autoStartup);
+			if (this.applicationEventPublisher != null) {
+				this.leaderInitiator.setLeaderEventPublisher(
+						new DefaultLeaderEventPublisher(this.applicationEventPublisher));
+			}
+		}
+	}
+
+	@Override
+	public synchronized LeaderInitiator getObject() throws Exception {
+		return this.leaderInitiator;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return LeaderInitiator.class;
+	}
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+
+}

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/CuratorFrameworkFactoryBeanTests.java

@@ -0,0 +1,45 @@
+/*
+ * 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.zookeeper.config;
+
+import static org.junit.Assert.assertTrue;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.test.TestingServer;
+import org.junit.Test;
+
+import org.springframework.integration.zookeeper.config.CuratorFrameworkFactoryBean;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class CuratorFrameworkFactoryBeanTests {
+
+	@Test
+	public void test() throws Exception {
+		TestingServer testingServer = new TestingServer();
+		CuratorFrameworkFactoryBean fb = new CuratorFrameworkFactoryBean(testingServer.getConnectString());
+		CuratorFramework client = fb.getObject();
+		fb.start();
+		assertTrue(client.getState().equals(CuratorFrameworkState.STARTED));
+		fb.stop();
+		assertTrue(client.getState().equals(CuratorFrameworkState.STOPPED));
+		testingServer.close();
+	}
+
+}

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/config/LeaderInitiatorFactoryBeanTests.java

@@ -0,0 +1,112 @@
+/*
+ * 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.zookeeper.config;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.junit.BeforeClass;
+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.event.AbstractLeaderEvent;
+import org.springframework.integration.leader.event.OnGrantedEvent;
+import org.springframework.integration.leader.event.OnRevokedEvent;
+import org.springframework.integration.zookeeper.ZookeeperTestSupport;
+import org.springframework.integration.zookeeper.leader.LeaderInitiator;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class LeaderInitiatorFactoryBeanTests extends ZookeeperTestSupport {
+
+	private static CuratorFramework client;
+
+	@Autowired
+	private LeaderInitiator leaderInitiator;
+
+	@Autowired
+	private Config config;
+
+	@BeforeClass
+	public static void getClient() throws Exception {
+		client = createNewClient();
+	}
+
+	@Test
+	public void test() throws Exception {
+		assertTrue(this.config.latch1.await(10, TimeUnit.SECONDS));
+		assertThat(this.config.events.get(0), instanceOf(OnGrantedEvent.class));
+		this.leaderInitiator.stop();
+		assertTrue(this.config.latch2.await(10, TimeUnit.SECONDS));
+		assertThat(this.config.events.get(1), instanceOf(OnRevokedEvent.class));
+	}
+
+	@Configuration
+	public static class Config {
+
+		private final List<AbstractLeaderEvent> events = new ArrayList<AbstractLeaderEvent>();
+
+		private final CountDownLatch latch1 = new CountDownLatch(1);
+
+		private final CountDownLatch latch2 = new CountDownLatch(2);
+
+		@Bean
+		public LeaderInitiatorFactoryBean leaderInitiator(CuratorFramework client) {
+			return new LeaderInitiatorFactoryBean(client, "/siTest/", "foo");
+		}
+
+		@Bean
+		public CuratorFramework client() {
+			return LeaderInitiatorFactoryBeanTests.client;
+		}
+
+		@Bean
+		public ApplicationListener<?> listener() {
+			return new ApplicationListener<AbstractLeaderEvent>() {
+
+				@Override
+				public void onApplicationEvent(AbstractLeaderEvent event) {
+					events.add(event);
+					latch1.countDown();
+					latch2.countDown();
+				}
+
+			};
+		}
+
+	}
+
+}

src/reference/asciidoc/aggregator.adoc

@@ -235,6 +235,14 @@ This implementation returns the value of one of the message headers (whose name
 By default, the correlation strategy is a `HeaderAttributeCorrelationStrategy` returning the value of the CORRELATION_ID header attribute.
 If you have a custom header name you would like to use for correlation, then simply configure that on an instance of `HeaderAttributeCorrelationStrategy` and provide that as a reference for the Aggregator's correlation-strategy.
 
+===== LockRegistry
+
+Changes to groups are thread safe; a `LockRegistry` is used to obtain a lock for the resolved correlation id.
+A `DefaultLockRegistry` is used by default (in-memory).
+For synchronizing updates across servers, where a shared `MessageGroupStore` is being used, a shared lock registry
+must be configured.
+See <<aggregator-config>> below for more information.
+
 [[aggregator-config]]
 ==== Configuring an Aggregator
 
@@ -426,8 +434,8 @@ Note that the actual time to expire an empty group will also be affected by the
 
 <20> A reference to a `org.springframework.integration.util.LockRegistry` bean; used to obtain a `Lock` based on the `groupId` for concurrent operations on the `MessageGroup`.
 By default, an internal `DefaultLockRegistry` is used.
-Use of a distributed `LockRegistry`, such as the `RedisLockRegistry`, ensures only one instance of the aggregator will operate on a group concurrently.
-See <<redis-lock-registry>> for more information.
+Use of a distributed `LockRegistry`, such as the `ZookeeperLockRegistry`, ensures only one instance of the aggregator will operate on a group concurrently.
+See <<redis-lock-registry>>, <<gemfire-lock-registry>>, <<zk-lock-registry>> for more information.
 
 
 

src/reference/asciidoc/endpoint.adoc

@@ -564,3 +564,62 @@ If you use an inner bean definition such as this:
 the bean is treated like any inner bean declared that way and is not registered with the application context.
 If you wish to access this bean in some other manner, declare it at the top level with an `id` and use the `ref` attribute instead.
 See the http://docs.spring.io/spring-framework/docs/current/spring-framework-reference/html/beans.html#beans-inner-beans[Spring Documentation] for more information.
+
+[[endpoint-roles]]
+=== Endpoint Roles
+
+Starting with _version 4.2_, endpoints can be assigned to roles.
+Roles allow endpoints to be started and stopped as a group; this is particularly useful when using leadership election
+where a set of endpoints can be started or stopped when leadership is granted or revoked respectively.
+
+You can assign endpoints to roles using XML, Java configuration, or programmatically:
+
+[source, xml]
+----
+<int:inbound-channel-adapter id="ica" channel="someChannel" expression="'foo'" role="cluster">
+    <int:poller fixed-rate="60000" />
+</int:inbound-channel-adapter>
+----
+
+[source, java]
+----
+@Bean
+@ServiceActivator(inputChannel = "sendAsyncChannel")
+@Role("cluster")
+public MessageHandler sendAsyncHandler() {
+    return // some MessageHandler
+}
+----
+
+[source, java]
+----
+@Payload("#args[0].toLowerCase()")
+@Role("cluster")
+public String handle(String payload) {
+    return payload.toUpperCase();
+}
+----
+
+[source, java]
+----
+@Autowired
+private SmartLifecycleRoleController roleController;
+
+...
+
+    this.roleController.addSmartLifeCycleToRole("cluster", someEndpoint);
+...
+----
+
+Each of these adds the endpoint to the role `cluster`.
+
+Invoking `roleController.startLifecyclesInRole("cluster")` (and the corresponding `stop...` method) will start/stop
+the endpoints.
+
+NOTE: Any object implementing `SmartLifecycle` can be programmatically added, not just endpoints.
+
+The `SmartLifecycleRoleController` implements `ApplicationListener<AbstractLeaderEvent>` and it will automatically
+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.

src/reference/asciidoc/index.adoc

@@ -103,6 +103,8 @@ include::./ws.adoc[]
 include::./xml.adoc[]
 
 include::./xmpp.adoc[]
+
+include::./zookeeper.adoc[]
 // BE SURE TO PRECEDE ALL include:: with a blank line - see https://github.com/asciidoctor/asciidoctor/issues/1297
 
 [[spring-integration-appendices]]

src/reference/asciidoc/meta-data-store.adoc

@@ -17,6 +17,7 @@ the framework:
 * <<redis-metadata-store>>
 * <<gemfire-metadata-store>>
 * <<mongodb-metadata-store>>
+* <<zk-metadata-store>>
 
 
 
@@ -61,3 +62,23 @@ The following configuration is an example of how to do this:
 The `value` of the idempotent entry may be some expiration date, after which that entry should be removed from _Metadata Store_ by some scheduled reaper.
 
 Also see <<idempotent-receiver>>.
+
+[[metadatastore-listener]]
+==== MetadataStoreListener
+
+Some metadata stores (currently only zookeeper) support registering a listener to receive events when items change.
+
+[source, java]
+----
+public interface MetadataStoreListener {
+
+	void onAdd(String key, String value);
+
+	void onRemove(String key, String oldValue);
+
+	void onUpdate(String key, String newValue);
+}
+----
+
+See the javadocs for more information.
+The `MetadataStoreListenerAdapter` can be subclassed if you are only interested in a subset of events.

src/reference/asciidoc/whats-new.adoc

@@ -35,6 +35,17 @@ For more information, see <<security>>.
 The `FileSplitter`, which splits text files into lines, was added in 4.1.2.
 It now has full support in the `int-file:` namespace; see <<file-splitter>> for more information.
 
+[[x4.2-zk]]
+==== Zookeeper Support
+
+Zookeeper support has been added to the framework to assist when running on a clustered/multi-host environment.
+
+* ZookeeperMetadataStore
+* ZookeeperLockRegistry
+* Zookeeper Leadership
+
+See <<zookeeper>> for more information.
+
 [[x4.2-general]]
 === General Changes
 

src/reference/asciidoc/zookeeper.adoc

@@ -0,0 +1,90 @@
+[[zookeeper]]
+== Zookeeper Support
+
+=== Introduction
+
+https://zookeeper.apache.org/[Zookeeper] support was added to the framework in _version 4.2_, comprised of:
+
+* MetadataStore
+* LockRegistry
+* Leadership Event Handling
+
+[[zk-metadata-store]]
+=== Zookeeper Metadata Store
+
+The `ZookeeperMetadataStore` can be used where any `MetadataStore` is needed, such as peristent file list filters,
+etc.
+See <<metadata-store>> for more information.
+
+[source, xml]
+----
+<bean id="client" class="org.springframework.integration.zookeeper.config.CuratorFrameworkFactoryBean">
+    <constructor-arg value="${connect.string}" />
+</bean>
+
+<bean id="meta" class="org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore">
+    <constructor-arg ref="client" />
+</bean>
+----
+
+[source, java]
+----
+@Bean
+public MetadataStore zkStore(CuratorFramework client) {
+    return new ZookeeperMetadataStore(client);
+}
+----
+
+[[zk-lock-registry]]
+=== Zookeeper Lock Registry
+
+The `ZookeeperLockRegistry` can be used where any `LockRegistry` is needed, such as when using an `Aggregator` in a
+clustered environment, with a shared `MessageStore`.
+
+A `LocRegistry` is used to "look up" a lock based on a key (the aggregator uses the `correlationId`).
+By default, locks in the `ZookeeperLockRegistry` are maintained in zookeeper under the path
+`/SpringIntegration-LockRegistry/`.
+You can customize the path by providing an implementation of `ZookeeperLockRegistry.KeyToPathStrategy`.
+
+[source, java]
+----
+public interface KeyToPathStrategy {
+
+    String pathFor(String key);
+
+    boolean bounded();
+
+}
+----
+
+If the strategy returns `true` from `isBounded`, unused locks do not need to be harvested.
+For unbounded strategies (such as the default) you will need to invoke `expireUnusedOlderThan(long age)` from time
+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>>).
+
+[source, xml]
+----
+<int-zk:leader-listener client="client" path="/siNamespace" role="cluster" />
+----
+
+`client` is a reference to a `CuratorFramework` bean; a `CuratorFrameworkFactoryBean` is available.
+When a leader is elected, an `OnGrantedEvent` will be published for the role `cluster`; any endpoints in that role
+will be started.
+When leadership is revoked, an `OnRevokedEvent` will be published for the role `cluster`; any endpoints in that role
+will be stopped.
+See <<endpoint-roles>> for more information.
+
+[source, java]
+----
+@Bean
+public LeaderInitiatorFactoryBean leaderInitiator(CuratorFramework client) {
+    return new LeaderInitiatorFactoryBean(client, "/siTest/", "cluster");
+}
+----