INT-3691 Add ZookeeperMetadataStore

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

- create new core interfaces `ListenableMetadataStore` and `MessageStoreListener`;
- create new module `spring-integration-zookeeper`;
- add `ZookeeperMetadataStore` implementation

Polishing

Polishing

- moved classes into core/test where necessary, including conversion methods
- properly renamed `MetadataStoreListenerAdapter`

build.gradle

@@ -86,6 +86,7 @@ subprojects { subproject ->
 		commonsDbcpVersion = '1.4'
 		commonsIoVersion = '2.4'
 		commonsNetVersion = '3.3'
+		curatorVersion = '2.8.0'
 		derbyVersion = '10.11.1.1'
 		eclipseLinkVersion = '2.4.2'
 		ftpServerVersion = '1.0.6'
@@ -135,6 +136,7 @@ subprojects { subproject ->
 		springWsVersion = '2.2.1.RELEASE'
 		xmlUnitVersion = '1.5'
 		xstreamVersion = '1.4.7'
+		zookeeperVersion = '3.4.6'
 	}
 
 	eclipse {
@@ -681,6 +683,22 @@ project('spring-integration-xmpp') {
 	}
 }
 
+project('spring-integration-zookeeper') {
+	description = 'Spring Integration Zookeeper Support'
+	dependencies {
+		compile project(":spring-integration-core")
+		compile ("org.apache.zookeeper:zookeeper:$zookeeperVersion")
+
+		compile("org.apache.curator:curator-recipes:$curatorVersion") {
+			exclude group: 'org.apache.zookeeper', module: 'zookeeper'
+			exclude group: 'org.jboss.netty', module: 'netty'
+		}
+
+		testCompile "org.springframework.integration:spring-integration-test"
+		testCompile "org.apache.curator:curator-test:$curatorVersion"
+	}
+}
+
 project("spring-integration-bom") {
 	description = "Spring Integration (Bill of Materials)"
 

spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2002-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.metadata;
+
+/**
+ *
+ * {@link ConcurrentMetadataStore} with the ability of registering {@link MetadataStoreListener} callbacks, to be
+ * invoked when changes occur in the metadata store.
+ *
+ * @author Marius Bogoevici
+ */
+public interface ListenableMetadataStore extends ConcurrentMetadataStore {
+
+	/**
+	 * Registers a listener with the metadata store
+	 *
+	 * @param callback the callback to be registered
+	 */
+	void addListener(MetadataStoreListener callback);
+
+	/**
+	 * Unregisters a listener
+	 *
+	 * @param callback the callback to be unregistered
+	 */
+	void removeListener(MetadataStoreListener callback);
+}

spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java

@@ -0,0 +1,48 @@
+/*
+ * 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.metadata;
+
+/**
+ * A callback to be invoked whenever a value changes in the data store.
+ *
+ * @author Marius Bogoevici
+ */
+public interface MetadataStoreListener {
+
+	/**
+	 * Invoked when a key is added to the store
+	 *
+	 * @param key the key being added
+	 * @param value the value being added
+	 */
+	void onAdd(String key, String value);
+
+	/**
+	 * Invoked when a key is removed from the store
+	 *
+	 * @param key the key being removed
+	 * @param oldValue the value being removed
+	 */
+	void onRemove(String key, String oldValue);
+
+	/**
+	 * Invoked when a key is updated into the store
+	 * @param key the key being updated
+	 * @param newValue the new value
+	 */
+	void onUpdate(String key, String newValue);
+}

spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListenerAdapter.java

@@ -0,0 +1,40 @@
+/*
+ * 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.metadata;
+
+/**
+ * Base implementation for a {@link MetadataStoreListener}. Subclasses may override any of the methods.
+ *
+ * @author Marius Bogoevici
+ */
+public abstract class MetadataStoreListenerAdapter implements MetadataStoreListener {
+
+	@Override
+	public void onAdd(String key, String value) {
+
+	}
+
+	@Override
+	public void onRemove(String key, String oldValue) {
+
+	}
+
+	@Override
+	public void onUpdate(String key, String newValue) {
+
+	}
+}

spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * 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.
@@ -15,6 +15,8 @@
  */
 package org.springframework.integration.support.utils;
 
+import java.io.UnsupportedEncodingException;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -23,11 +25,13 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * General utility methods.
  *
  * @author Gary Russell
+ * @author Marius Bogoevici
  * @since 4.0
  *
  */
@@ -98,4 +102,37 @@ public class IntegrationUtils {
 		return beanFactory.getBean(beanName, type);
 	}
 
+	/**
+	 * Utility method for null-safe conversion from String to byte[]
+	 *
+	 * @param value the String to be converted
+	 * @param encoding the encoding
+	 * @return the byte[] corresponding to the given String and encoding, null if provided String argument was null
+	 * @throws IllegalArgumentException if the encoding is not supported
+	 */
+	public static byte[] stringToBytes(String value, String encoding) {
+		try {
+			return value != null ? value.getBytes(encoding) : null;
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new IllegalArgumentException(e);
+		}
+	}
+
+	/**
+	 * Utility method for null-safe conversion from byte[] to String
+	 *
+	 * @param bytes the byte[] to be converted
+	 * @param encoding the encoding
+	 * @return the String corresponding to the given byte[] and encoding, null if provided byte[] argument was null
+	 * @throws IllegalArgumentException if the encoding is not supported
+	 */
+	public static String bytesToString(byte[] bytes, String encoding) {
+		try {
+			return bytes == null ? null : new String(bytes, encoding);
+		}
+		catch (UnsupportedEncodingException e) {
+			throw new IllegalArgumentException(e);
+		}
+	}
 }

spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java

@@ -0,0 +1,57 @@
+/*
+ * 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.test.matcher;
+
+import org.hamcrest.Description;
+import org.hamcrest.DiagnosingMatcher;
+
+import org.springframework.util.ObjectUtils;
+
+/**
+ * A matcher that evaluates against the result of invoking a function,
+ * wrapped by the {@link EqualsResultMatcher.Evaluator}
+ *
+ * The goal is to defer the computation until the matcher needs to be actually evaluated.
+ * Mainly useful in conjunction with retrying matcherss such as {@link EventuallyMatcher}
+ *
+ * @author Marius Bogoevici
+ */
+public class EqualsResultMatcher<U> extends DiagnosingMatcher<U> {
+
+	private Evaluator<U> evaluator;
+
+	public EqualsResultMatcher(Evaluator<U> evaluator) {
+		this.evaluator = evaluator;
+	}
+
+	@Override
+	protected boolean matches(Object item, Description mismatchDescription) {
+		return ObjectUtils.nullSafeEquals(item, evaluator.evaluate());
+	}
+
+	@Override
+	public void describeTo(Description description) {
+	}
+
+	public interface Evaluator<U> {
+		U evaluate();
+	}
+
+	public static <U> EqualsResultMatcher<U> equalsResult(Evaluator<U> evaluator) {
+		return new EqualsResultMatcher<U>(evaluator);
+	}
+}

spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2013-2104 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.test.matcher;
+
+import org.hamcrest.Description;
+import org.hamcrest.DiagnosingMatcher;
+import org.hamcrest.Matcher;
+
+
+/**
+ * A matcher that will evaluate another matcher repeatedly until it matches, or fail after some number of attempts.
+ *
+ * @param <U> the type the wrapped matcher operates on
+ *
+ * (Copied from {@code org.springframework.xd.test.fixtures.EventuallyMatcher})
+ *
+ * @author Eric Bottard
+ */
+public class EventuallyMatcher<U> extends DiagnosingMatcher<U> {
+
+	private final Matcher<U> delegate;
+
+	private int nbAttempts;
+
+	private int pause;
+
+	public EventuallyMatcher(Matcher<U> delegate) {
+		this(delegate, 20, 100);
+	}
+
+	public EventuallyMatcher(Matcher<U> delegate, int nbAttempts, int pause) {
+		this.delegate = delegate;
+		this.nbAttempts = nbAttempts;
+		this.pause = pause;
+	}
+
+	public static <U> Matcher<U> eventually(int nbAttempts, int pause, Matcher<U> delegate) {
+		return new EventuallyMatcher<U>(delegate,nbAttempts,pause);
+	}
+
+	public static <U> Matcher<U> eventually(Matcher<U> delegate) {
+		return new EventuallyMatcher<U>(delegate);
+	}
+
+	@Override
+	public void describeTo(Description description) {
+		description.appendDescriptionOf(delegate).appendText(String.format(", trying at most %d times", nbAttempts));
+	}
+
+	@Override
+	protected boolean matches(Object item, Description mismatchDescription) {
+		mismatchDescription.appendText(String.format("failed after %d*%d=%dms:%n", nbAttempts, pause, nbAttempts
+				* pause));
+		for (int i = 0; i < nbAttempts; i++) {
+			boolean result = delegate.matches(item);
+			if (result) {
+				return true;
+			}
+			delegate.describeMismatch(item, mismatchDescription);
+			mismatchDescription.appendText(", ");
+			try {
+				Thread.sleep(pause);
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				break;
+			}
+		}
+		return false;
+	}
+}

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java

@@ -0,0 +1,390 @@
+/*
+ * 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.metadata;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.utils.EnsurePath;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.Stat;
+
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.metadata.ListenableMetadataStore;
+import org.springframework.integration.metadata.MetadataStoreListener;
+import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.util.Assert;
+
+/**
+ * Zookeeper-based {@link ListenableMetadataStore} based on a Zookeeper node. Values are stored in the children node,
+ * the names of which are stored as keys.
+ *
+ * @author Marius Bogoevici
+ */
+public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLifecycle {
+
+	private final Object lifecycleMonitor = new Object();
+
+	private final CuratorFramework client;
+
+	private final List<MetadataStoreListener> listeners = new CopyOnWriteArrayList<MetadataStoreListener>();
+
+	/**
+	 * An internal map storing local updates, ensuring that they have precedence if the cache contains stale data.
+	 * As changes are propagated back from Zookeeper to the cache, entries are removed.
+	 */
+	private final ConcurrentMap<String, LocalChildData> updateMap = new ConcurrentHashMap<String,LocalChildData>();
+
+	private volatile String root = "/SpringIntegration-MetadataStore";
+
+	private volatile String encoding = "UTF-8";
+
+	private volatile PathChildrenCache cache;
+
+	private volatile boolean running = false;
+
+	private volatile boolean autoStartup = true;
+
+	private volatile int phase = Integer.MAX_VALUE;
+
+	public ZookeeperMetadataStore(CuratorFramework client) throws Exception {
+		Assert.notNull(client, "Client cannot be null");
+		this.client = client;
+	}
+
+	/**
+	 * Encoding to use when storing data in ZooKeeper
+	 *
+	 * @param encoding encoding as text
+	 */
+	public void setEncoding(String encoding) {
+		Assert.hasText(encoding, "'encoding' cannot be null or empty.");
+		this.encoding = encoding;
+	}
+
+	/**
+	 * Root node - store entries are children of this node.
+	 *
+	 * @param root encoding as text
+	 */
+	public void setRoot(String root) {
+		Assert.notNull(root, "'root' must not be null.");
+		Assert.isTrue(root.startsWith("/"), "'root' must start with '/'");
+		// remove trailing slash, if not root
+		this.root = "/".equals(root) || !root.endsWith("/") ? root : root.substring(0, root.length() - 1);
+	}
+
+	public String getRoot() {
+		return this.root;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(value, "'value' must not be null.");
+		synchronized (this.updateMap) {
+			try {
+				createNode(key, value);
+				return null;
+			}
+			catch (KeeperException.NodeExistsException e) {
+				// so the data actually exists, we can read it
+				try {
+					byte[] bytes = this.client.getData().forPath(getPath(key));
+					return IntegrationUtils.bytesToString(bytes, encoding);
+				}
+				catch (Exception exceptionDuringGet) {
+					throw new ZookeeperMetadataStoreException("Exception while reading node with key '" + key + "':", e);
+				}
+			}
+			catch (Exception e) {
+				throw new ZookeeperMetadataStoreException("Error while trying to set '" + key + "':", e);
+			}
+		}
+	}
+
+	@Override
+	public boolean replace(String key, String oldValue, String newValue) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(oldValue, "'oldValue' must not be null.");
+		Assert.notNull(newValue, "'newValue' must not be null.");
+		synchronized (this.updateMap) {
+			Stat currentStat = new Stat();
+			try {
+				byte[] bytes = this.client.getData().storingStatIn(currentStat).forPath(getPath(key));
+				if (oldValue.equals(IntegrationUtils.bytesToString(bytes, encoding))) {
+					updateNode(key, newValue, currentStat.getVersion());
+				}
+				return true;
+			}
+			catch (KeeperException.NoNodeException e) {
+				// ignore, the node doesn't exist there's nothing to replace
+				return false;
+			}
+			catch (KeeperException.BadVersionException e) {
+				// ignore
+				return false;
+			}
+			catch (Exception e) {
+				throw new ZookeeperMetadataStoreException("Cannot replace value");
+			}
+		}
+	}
+
+	@Override
+	public void addListener(MetadataStoreListener callback) {
+		this.listeners.add(callback);
+	}
+
+	@Override
+	public void removeListener(MetadataStoreListener callback) {
+		this.listeners.remove(callback);
+	}
+
+	@Override
+	public void put(String key, String value) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(value, "'value' must not be null.");
+		synchronized (this.updateMap) {
+			try {
+				Stat currentNode = this.client.checkExists().forPath(getPath(key));
+				if (currentNode == null) {
+					try {
+						createNode(key, value);
+					}
+					catch (KeeperException.NodeExistsException e) {
+						updateNode(key, value, -1);
+					}
+				}
+				else {
+					updateNode(key, value, -1);
+				}
+			}
+			catch (Exception e) {
+				throw new ZookeeperMetadataStoreException("Error while setting value for key '" + key + "':", e);
+			}
+		}
+	}
+
+	@Override
+	public String get(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		synchronized (this.updateMap) {
+			ChildData currentData = this.cache.getCurrentData(getPath(key));
+			if (currentData == null) {
+				if (this.updateMap.containsKey(key)) {
+					// we have saved the value, but the cache hasn't updated yet
+					// if the value had changed via replication, we would have been notified by the listener
+					return this.updateMap.get(key).getValue();
+				}
+				else {
+					// the value just doesn't exist
+					return null;
+				}
+			}
+			else {
+				if (this.updateMap.containsKey(key)) {
+					// our version is more recent than the cache
+					if (this.updateMap.get(key).getVersion() >= currentData.getStat().getVersion()) {
+						return this.updateMap.get(key).getValue();
+					}
+				}
+				return IntegrationUtils.bytesToString(currentData.getData(), encoding);
+			}
+		}
+	}
+
+	@Override
+	public String remove(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		synchronized (this.updateMap) {
+			try {
+				byte[] bytes = this.client.getData().forPath(getPath(key));
+				this.client.delete().forPath(getPath(key));
+				// we guarantee that the deletion will supersede the existing data
+				this.updateMap.put(key, new LocalChildData(null, Integer.MAX_VALUE));
+				return IntegrationUtils.bytesToString(bytes, encoding);
+			}
+			catch (KeeperException.NoNodeException e) {
+				// ignore - the node doesn't exist
+				return null;
+			}
+			catch (Exception e) {
+				throw new ZookeeperMetadataStoreException("Exception while deleting key '" + key + "'", e);
+			}
+		}
+	}
+
+	private void updateNode(String key, String value, int version) throws Exception {
+		Stat stat = this.client.setData().withVersion(version).forPath(getPath(key),
+				IntegrationUtils.stringToBytes(value, this.encoding));
+		this.updateMap.put(key, new LocalChildData(value, stat.getVersion()));
+	}
+
+	private void createNode(String key, String value) throws Exception {
+		this.client.create().forPath(getPath(key), IntegrationUtils.stringToBytes(value, this.encoding));
+		this.updateMap.put(key, new LocalChildData(value, 0));
+	}
+
+	public String getPath(String key) {
+		return "".equals(key) ? this.root : this.root + "/" + key;
+	}
+
+
+	@Override
+	public boolean isAutoStartup() {
+		return autoStartup;
+	}
+
+	@Override
+	public void start() {
+		if (!this.running) {
+			synchronized (this.lifecycleMonitor) {
+				if (!this.running) {
+					try {
+						EnsurePath ensurePath = new EnsurePath(root);
+						ensurePath.ensure(client.getZookeeperClient());
+						this.cache = new PathChildrenCache(this.client, this.root, true);
+						this.cache.getListenable().addListener(new MetadataStoreListenerInvokingPathChildrenCacheListener());
+						this.cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
+						running = true;
+					}
+					catch (Exception e) {
+						throw new ZookeeperMetadataStoreException("Exception while starting bean", e);
+					}
+				}
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.running) {
+			synchronized (this.lifecycleMonitor) {
+				if (this.running) {
+					if (this.cache != null) {
+						CloseableUtils.closeQuietly(this.cache);
+					}
+					this.cache = null;
+					this.running = false;
+				}
+			}
+		}
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return running;
+	}
+
+	@Override
+	public int getPhase() {
+		return phase;
+	}
+
+	private static class LocalChildData {
+
+		private String value;
+
+		private int version;
+
+		public LocalChildData(String value, int version) {
+			this.value = value;
+			this.version = version;
+		}
+
+		public String getValue() {
+			return this.value;
+		}
+
+		public int getVersion() {
+			return this.version;
+		}
+	}
+
+	private class MetadataStoreListenerInvokingPathChildrenCacheListener implements
+			PathChildrenCacheListener {
+		@Override
+		public void childEvent(CuratorFramework client, PathChildrenCacheEvent event)
+				throws Exception {
+			synchronized (ZookeeperMetadataStore.this.updateMap) {
+				String eventPath = event.getData().getPath();
+				String eventKey = getKey(eventPath);
+				byte[] eventData = event.getData().getData();
+				switch (event.getType()) {
+				case CHILD_ADDED:
+					if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) {
+						if (event.getData().getStat().getVersion() >=
+								ZookeeperMetadataStore.this.updateMap.get(eventKey).getVersion()) {
+							ZookeeperMetadataStore.this.updateMap.remove(eventPath);
+						}
+					}
+					for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
+						listener.onAdd(eventKey, IntegrationUtils.bytesToString(eventData, encoding));
+					}
+					break;
+				case CHILD_UPDATED:
+					if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) {
+						if (event.getData().getStat().getVersion() >=
+								ZookeeperMetadataStore.this.updateMap.get(eventKey).getVersion()) {
+							ZookeeperMetadataStore.this.updateMap.remove(eventPath);
+						}
+					}
+					for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
+						listener.onUpdate(eventKey, IntegrationUtils.bytesToString(eventData, encoding));
+					}
+					break;
+				case CHILD_REMOVED:
+					ZookeeperMetadataStore.this.updateMap.remove(eventKey);
+					for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) {
+						listener.onRemove(eventKey, IntegrationUtils.bytesToString(eventData, encoding));
+					}
+					break;
+				default:
+					// ignore all other events
+					break;
+				}
+			}
+		}
+	}
+
+	private String getKey(String path) {
+		return path.replace(root + "/", "");
+	}
+}

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java

@@ -0,0 +1,33 @@
+/*
+ * 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.metadata;
+
+/**
+ * @author Marius Bogoevici
+ */
+@SuppressWarnings("serial")
+public class ZookeeperMetadataStoreException extends RuntimeException {
+
+	public ZookeeperMetadataStoreException(String message) {
+		super(message);
+	}
+
+	public ZookeeperMetadataStoreException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+}

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * Provides classes supporting the Zookeeper-based
+ * {@link org.springframework.integration.metadata.ListenableMetadataStore}
+ */
+package org.springframework.integration.zookeeper.metadata;

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java

@@ -0,0 +1,422 @@
+/*
+ * 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.metadata;
+
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+import static org.springframework.integration.test.matcher.EqualsResultMatcher.equalsResult;
+import static org.springframework.integration.test.matcher.EventuallyMatcher.eventually;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.BoundedExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.apache.curator.utils.CloseableUtils;
+import org.hamcrest.collection.IsIterableContainingInOrder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.integration.metadata.MetadataStoreListenerAdapter;
+import org.springframework.integration.metadata.MetadataStoreListener;
+import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.integration.test.matcher.EqualsResultMatcher.Evaluator;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class ZookeeperMetadataStoreTests {
+
+	private static final Log log = LogFactory.getLog(ZookeeperMetadataStore.class);
+
+	private TestingServer testingServer;
+
+	private CuratorFramework client;
+
+	private ZookeeperMetadataStore metadataStore;
+
+	@Before
+	public void setUp() throws Exception{
+		testingServer = new TestingServer(true);
+		client = createNewClient();
+		metadataStore = new ZookeeperMetadataStore(client);
+		metadataStore.start();
+	}
+
+	@After
+	public void tearDown() {
+		this.metadataStore.stop();
+		closeClient(this.client);
+		try {
+			testingServer.stop();
+		}
+		catch (IOException e) {
+			log.warn("Exception thrown while shutting down ZooKeeper: ", e);
+		}
+		testingServer.getTempDirectory().delete();
+	}
+
+
+	@Test
+	public void testGetNonExistingKeyValue() {
+		String retrievedValue = metadataStore.get("does-not-exist");
+		assertNull(retrievedValue);
+	}
+
+	@Test
+	public void testPersistKeyValue() throws Exception {
+		String testKey = "ZookeeperMetadataStoreTests-Persist";
+		metadataStore.put(testKey, "Integration");
+		assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey)));
+		assertEquals("Integration",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+	}
+
+
+	@Test
+	public void testGetValueFromMetadataStore() throws Exception {
+		String testKey = "ZookeeperMetadataStoreTests-GetValue";
+		metadataStore.put(testKey, "Hello Zookeeper");
+		String retrievedValue = metadataStore.get(testKey);
+		assertEquals("Hello Zookeeper", retrievedValue);
+	}
+
+
+	@Test
+	public void testPutIfAbsent() throws Exception {
+		final String testKey = "ZookeeperMetadataStoreTests-Persist";
+		final String testKey2 = "ZookeeperMetadataStoreTests-Persist-2";
+		metadataStore.put(testKey, "Integration");
+		assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey)));
+		assertEquals("Integration",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+		CuratorFramework otherClient = createNewClient();
+		final ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient);
+		otherMetadataStore.start();
+		otherMetadataStore.putIfAbsent(testKey, "OtherValue");
+		assertEquals("Integration",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+		assertEquals("Integration", metadataStore.get(testKey));
+		assertThat("Integration", eventually(equalsResult(new Evaluator<String>() {
+			@Override
+			public String evaluate() {
+				return otherMetadataStore.get(testKey);
+			}
+		})));
+		otherMetadataStore.putIfAbsent(testKey2, "Integration-2");
+		assertEquals("Integration-2",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey2)), "UTF-8"));
+		assertEquals("Integration-2", otherMetadataStore.get(testKey2));
+		assertThat("Integration-2", eventually(equalsResult(new Evaluator<String>() {
+			@Override
+			public String evaluate() {
+				return metadataStore.get(testKey2);
+			}
+		})));
+		closeClient(otherClient);
+	}
+
+	@Test
+	public void testReplace() throws Exception {
+		final String testKey = "ZookeeperMetadataStoreTests-Replace";
+		metadataStore.put(testKey, "Integration");
+		assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey)));
+		assertEquals("Integration",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+		CuratorFramework otherClient = createNewClient();
+		final ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient);
+		otherMetadataStore.start();
+		otherMetadataStore.replace(testKey, "OtherValue", "Integration-2");
+		assertEquals("Integration",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+		assertEquals("Integration", metadataStore.get(testKey));
+		assertThat("Integration", eventually(equalsResult(new Evaluator<String>() {
+			@Override
+			public String evaluate() {
+				return otherMetadataStore.get(testKey);
+			}
+		})));
+		otherMetadataStore.replace(testKey, "Integration", "Integration-2");
+		assertEquals("Integration-2",
+				IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8"));
+		assertThat("Integration-2", eventually(equalsResult(new Evaluator<String>() {
+			@Override
+			public String evaluate() {
+				return metadataStore.get(testKey);
+			}
+		})));
+		assertEquals("Integration-2", otherMetadataStore.get(testKey));
+		closeClient(otherClient);
+	}
+
+	@Test
+	public void testPersistEmptyStringToMetadataStore() {
+		String testKey = "ZookeeperMetadataStoreTests-PersistEmpty";
+		metadataStore.put(testKey, "");
+		assertEquals("", metadataStore.get(testKey));
+	}
+
+	@Test
+	public void testPersistNullStringToMetadataStore() {
+		try {
+			metadataStore.put("ZookeeperMetadataStoreTests-PersistEmpty", null);
+		}
+		catch (IllegalArgumentException e) {
+			assertEquals("'value' must not be null.", e.getMessage());
+			return;
+		}
+		fail("Expected an IllegalArgumentException to be thrown.");
+	}
+
+	@Test
+	public void testPersistWithEmptyKeyToMetadataStore() {
+		metadataStore.put("", "PersistWithEmptyKey");
+		String retrievedValue = metadataStore.get("");
+		assertEquals("PersistWithEmptyKey", retrievedValue);
+	}
+
+	@Test
+	public void testPersistWithNullKeyToMetadataStore() {
+		try {
+			metadataStore.put(null, "something");
+		}
+		catch (IllegalArgumentException e) {
+			assertEquals("'key' must not be null.", e.getMessage());
+			return;
+		}
+		fail("Expected an IllegalArgumentException to be thrown.");
+	}
+
+	@Test
+	public void testGetValueWithNullKeyFromMetadataStore() {
+		try {
+			metadataStore.get(null);
+		}
+		catch (IllegalArgumentException e) {
+			assertEquals("'key' must not be null.", e.getMessage());
+			return;
+		}
+		fail("Expected an IllegalArgumentException to be thrown.");
+	}
+
+	@Test
+	public void testRemoveFromMetadataStore() throws Exception {
+		String testKey = "ZookeeperMetadataStoreTests-Remove";
+		String testValue = "Integration";
+		metadataStore.put(testKey, testValue);
+		assertEquals(testValue, metadataStore.remove(testKey));
+		Thread.sleep(1000);
+		assertNull(metadataStore.remove(testKey));
+	}
+
+	@Test
+	public void testListenerInvokedOnLocalChanges() throws Exception {
+		String testKey = "ZookeeperMetadataStoreTests";
+
+		// register listeners
+		final List<List<String>> notifiedChanges = new ArrayList<List<String>>();
+		final Map<String,CyclicBarrier> barriers = new HashMap<String, CyclicBarrier>();
+		barriers.put("add", new CyclicBarrier(2));
+		barriers.put("remove", new CyclicBarrier(2));
+		barriers.put("update", new CyclicBarrier(2));
+		metadataStore.addListener(new MetadataStoreListenerAdapter() {
+			@Override
+			public void onAdd(String key, String value) {
+				notifiedChanges.add(Arrays.asList("add", key, value));
+				waitAtBarrier("add", barriers);
+			}
+
+			@Override
+			public void onRemove(String key, String oldValue) {
+				notifiedChanges.add(Arrays.asList("remove", key, oldValue));
+				waitAtBarrier("remove", barriers);
+			}
+
+			@Override
+			public void onUpdate(String key, String newValue) {
+				notifiedChanges.add(Arrays.asList("update", key, newValue));
+				waitAtBarrier("update", barriers);
+			}
+		});
+
+		// the tests themselves
+		barriers.get("add").reset();
+		metadataStore.put(testKey, "Integration");
+		waitAtBarrier("add", barriers);
+		assertThat(notifiedChanges, hasSize(1));
+		assertThat(notifiedChanges.get(0), IsIterableContainingInOrder.contains("add", testKey, "Integration"));
+
+		metadataStore.putIfAbsent(testKey, "Integration++");
+		// there is no update and therefore we expect no changes
+		assertThat(notifiedChanges, hasSize(1));
+
+		barriers.get("update").reset();
+		metadataStore.put(testKey, "Integration-2");
+		waitAtBarrier("update", barriers);
+		assertThat(notifiedChanges, hasSize(2));
+		assertThat(notifiedChanges.get(1), IsIterableContainingInOrder.contains("update", testKey, "Integration-2"));
+
+		barriers.get("update").reset();
+		metadataStore.replace(testKey, "Integration-2", "Integration-3");
+		waitAtBarrier("update", barriers);
+		assertThat(notifiedChanges, hasSize(3));
+		assertThat(notifiedChanges.get(2), IsIterableContainingInOrder.contains("update", testKey, "Integration-3"));
+
+		metadataStore.replace(testKey, "Integration-2", "Integration-none");
+		assertThat(notifiedChanges, hasSize(3));
+
+		barriers.get("remove").reset();
+		metadataStore.remove(testKey);
+		waitAtBarrier("remove", barriers);
+		assertThat(notifiedChanges, hasSize(4));
+		assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3"));
+
+		// sleep and try to see if there were any other updates
+		Thread.sleep(1000);
+		assertThat(notifiedChanges, hasSize(4));
+	}
+
+	@Test
+	public void testListenerInvokedOnRemoteChanges() throws Exception {
+		String testKey = "ZookeeperMetadataStoreTests";
+
+		CuratorFramework otherClient = createNewClient();
+		ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient);
+
+		// register listeners
+		final List<List<String>> notifiedChanges = new ArrayList<List<String>>();
+		final Map<String,CyclicBarrier> barriers = new HashMap<String, CyclicBarrier>();
+		barriers.put("add", new CyclicBarrier(2));
+		barriers.put("remove", new CyclicBarrier(2));
+		barriers.put("update", new CyclicBarrier(2));
+		metadataStore.addListener(new MetadataStoreListenerAdapter() {
+			@Override
+			public void onAdd(String key, String value) {
+				notifiedChanges.add(Arrays.asList("add", key, value));
+				waitAtBarrier("add", barriers);
+			}
+
+			@Override
+			public void onRemove(String key, String oldValue) {
+				notifiedChanges.add(Arrays.asList("remove", key, oldValue));
+				waitAtBarrier("remove", barriers);
+			}
+
+			@Override
+			public void onUpdate(String key, String newValue) {
+				notifiedChanges.add(Arrays.asList("update", key, newValue));
+				waitAtBarrier("update", barriers);
+			}
+		});
+
+		// the tests themselves
+		barriers.get("add").reset();
+		otherMetadataStore.put(testKey, "Integration");
+		waitAtBarrier("add", barriers);
+		assertThat(notifiedChanges, hasSize(1));
+		assertThat(notifiedChanges.get(0), IsIterableContainingInOrder.contains("add", testKey, "Integration"));
+
+		otherMetadataStore.putIfAbsent(testKey, "Integration++");
+		// there is no update and therefore we expect no changes
+		assertThat(notifiedChanges, hasSize(1));
+
+		barriers.get("update").reset();
+		otherMetadataStore.put(testKey, "Integration-2");
+		waitAtBarrier("update", barriers);
+		assertThat(notifiedChanges, hasSize(2));
+		assertThat(notifiedChanges.get(1), IsIterableContainingInOrder.contains("update", testKey, "Integration-2"));
+
+		barriers.get("update").reset();
+		otherMetadataStore.replace(testKey, "Integration-2", "Integration-3");
+		waitAtBarrier("update", barriers);
+		assertThat(notifiedChanges, hasSize(3));
+		assertThat(notifiedChanges.get(2), IsIterableContainingInOrder.contains("update", testKey, "Integration-3"));
+
+		otherMetadataStore.replace(testKey, "Integration-2", "Integration-none");
+		assertThat(notifiedChanges, hasSize(3));
+
+		barriers.get("remove").reset();
+		otherMetadataStore.remove(testKey);
+		waitAtBarrier("remove", barriers);
+		assertThat(notifiedChanges, hasSize(4));
+		assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3"));
+
+		// sleep and try to see if there were any other updates - if there any pending updates, we should catch them by now
+		Thread.sleep(1000);
+		assertThat(notifiedChanges, hasSize(4));
+	}
+
+	@Test
+	public void testAddRemoveListener() throws Exception {
+		MetadataStoreListener mockListener = Mockito.mock(MetadataStoreListener.class);
+		DirectFieldAccessor accessor = new DirectFieldAccessor(metadataStore);
+
+		@SuppressWarnings("unchecked")
+		List<MetadataStoreListener> listeners = (List<MetadataStoreListener>) accessor.getPropertyValue("listeners");
+
+		assertThat(listeners, hasSize(0));
+		metadataStore.addListener(mockListener);
+		assertThat(listeners, hasSize(1));
+		assertThat(listeners, IsIterableContainingInOrder.contains(mockListener));
+		metadataStore.removeListener(mockListener);
+		assertThat(listeners, hasSize(0));
+	}
+
+	private void waitAtBarrier(String barrierName, Map<String, CyclicBarrier> barriers) {
+		try {
+			barriers.get(barrierName).await(10, TimeUnit.SECONDS);
+		}
+		catch (Exception e) {
+			throw new AssertionError("Test didn't complete: ", e);
+		}
+	}
+
+	private CuratorFramework createNewClient() throws InterruptedException {
+		CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(),
+				new BoundedExponentialBackoffRetry(100, 1000, 3));
+		client.start();
+		client.blockUntilConnected(10000, TimeUnit.SECONDS);
+		return client;
+	}
+
+	private void closeClient(CuratorFramework client) {
+		try {
+			CloseableUtils.closeQuietly(client);
+		}
+		catch (Exception e) {
+			log.warn("Exception thrown while closing client: ", e);
+		}
+	}
+
+}

spring-integration-zookeeper/src/test/resources/log4j.properties

@@ -0,0 +1,8 @@
+log4j.rootCategory=WARN, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n
+
+log4j.category.org.springframework.integration=WARN
+log4j.category.org.springframework.integration.zookeeper=INFO