INT-3353 Concurrent RedisMetadataStore

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

Shared metadata for PersistentAcceptOnceFileListFilters.

INT-3353 Polishing; PR Comments

build.gradle

@@ -258,6 +258,9 @@ project('spring-integration-file') {
 		compile project(":spring-integration-core")
 		compile "org.springframework:spring-context:$springVersion"
 		compile "commons-io:commons-io:$commonsIoVersion"
+		testCompile project(":spring-integration-redis")
+		testCompile project(":spring-integration-redis").sourceSets.test.output
+		testCompile "com.lambdaworks:lettuce:$lettuceVersion"
 	}
 }
 

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2014 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;
+
+
+
+/**
+ * Supports atomic updates to values in the store.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public interface ConcurrentMetadataStore extends MetadataStore {
+
+	/**
+	 * Atomically insert the key into the store.
+	 *
+	 * @param key The key.
+	 * @param value The value.
+	 * @return null if successful, the old value otherwise.
+	 */
+	String putIfAbsent(String key, String value);
+
+	/**
+	 * Atomically replace the value for the key in the store if the old
+	 * value matches the oldValue argument.
+	 *
+	 * @param key The key.
+	 * @param oldValue The old value.
+	 * @param newValue The new value.
+	 * @return true if successful.
+	 */
+	boolean replace(String key, String oldValue, String newValue);
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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,12 +25,15 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.Properties;
+import java.util.concurrent.locks.Lock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.util.DefaultLockRegistry;
+import org.springframework.integration.util.LockRegistry;
 import org.springframework.util.Assert;
 import org.springframework.util.DefaultPropertiesPersister;
 
@@ -45,7 +48,7 @@ import org.springframework.util.DefaultPropertiesPersister;
  * @author Gary Russell
  * @since 2.0
  */
-public class PropertiesPersistingMetadataStore implements MetadataStore, InitializingBean, DisposableBean {
+public class PropertiesPersistingMetadataStore implements ConcurrentMetadataStore, InitializingBean, DisposableBean {
 
 	private final Log logger = LogFactory.getLog(getClass());
 
@@ -53,9 +56,11 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 
 	private final DefaultPropertiesPersister persister = new DefaultPropertiesPersister();
 
-	private volatile File file;
+	private final LockRegistry lockRegistry = new DefaultLockRegistry();
 
-	private volatile String baseDirectory = System.getProperty("java.io.tmpdir") + "/spring-integration/";
+	private String baseDirectory = System.getProperty("java.io.tmpdir") + "/spring-integration/";
+
+	private File file;
 
 
 	public void setBaseDirectory(String baseDirectory) {
@@ -82,17 +87,85 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 
 	@Override
 	public void put(String key, String value) {
-		this.metadata.setProperty(key, value);
+		Assert.notNull(key, "'key' cannot be null");
+		Assert.notNull(value, "'value' cannot be null");
+		Lock lock = this.lockRegistry.obtain(key);
+		lock.lock();
+		try {
+			this.metadata.setProperty(key, value);
+		}
+		finally {
+			lock.unlock();
+		}
 	}
 
 	@Override
 	public String get(String key) {
-		return this.metadata.getProperty(key);
+		Assert.notNull(key, "'key' cannot be null");
+		Lock lock = this.lockRegistry.obtain(key);
+		lock.lock();
+		try {
+			return this.metadata.getProperty(key);
+		}
+		finally {
+			lock.unlock();
+		}
 	}
 
 	@Override
 	public String remove(String key) {
-		return (String) this.metadata.remove(key);
+		Assert.notNull(key, "'key' cannot be null");
+		Lock lock = this.lockRegistry.obtain(key);
+		lock.lock();
+		try {
+			return (String) this.metadata.remove(key);
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		Assert.notNull(key, "'key' cannot be null");
+		Assert.notNull(value, "'value' cannot be null");
+		Lock lock = this.lockRegistry.obtain(key);
+		lock.lock();
+		try {
+			String property = this.metadata.getProperty(key);
+			if (property == null) {
+				this.metadata.setProperty(key, value);
+				return null;
+			}
+			else {
+				return property;
+			}
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public boolean replace(String key, String oldValue, String newValue) {
+		Assert.notNull(key, "'key' cannot be null");
+		Assert.notNull(oldValue, "'oldValue' cannot be null");
+		Assert.notNull(newValue, "'newValue' cannot be null");
+		Lock lock = this.lockRegistry.obtain(key);
+		lock.lock();
+		try {
+			String property = this.metadata.getProperty(key);
+			if (oldValue.equals(property)) {
+				this.metadata.setProperty(key, newValue);
+				return true;
+			}
+			else {
+				return false;
+			}
+		}
+		finally {
+			lock.unlock();
+		}
 	}
 
 	@Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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
@@ -13,8 +13,8 @@
 
 package org.springframework.integration.metadata;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 
 /**
@@ -22,24 +22,37 @@ import java.util.Map;
  * The metadata will not be persisted across application restarts.
  *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
-public class SimpleMetadataStore implements MetadataStore {
+public class SimpleMetadataStore implements ConcurrentMetadataStore {
 
-	private final Map<String, String> metadata = new HashMap<String, String>();
+	private final ConcurrentMap<String, String> metadata = new ConcurrentHashMap<String, String>();
 
 
+	@Override
 	public void put(String key, String value) {
 		this.metadata.put(key, value);
 	}
 
+	@Override
 	public String get(String key) {
 		return this.metadata.get(key);
 	}
 
 	@Override
 	public String remove(String key) {
-		return metadata.remove(key);
+		return this.metadata.remove(key);
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		return this.metadata.putIfAbsent(key, value);
+	}
+
+	@Override
+	public boolean replace(String key, String oldValue, String newValue) {
+		return this.metadata.replace(key, oldValue, newValue);
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -17,7 +17,9 @@
 package org.springframework.integration.metadata;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
@@ -32,6 +34,7 @@ import org.springframework.core.io.support.PropertiesLoaderUtils;
  * @author Oleg Zhurakousky
  * @author Mark Fisher
  * @author Gunnar Hillert
+ * @author Gary Russell
  * @since 2.0
  */
 public class PropertiesPersistingMetadataStoreTests {
@@ -43,7 +46,10 @@ public class PropertiesPersistingMetadataStoreTests {
 		PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
 		metadataStore.afterPropertiesSet();
 		assertTrue(file.exists());
-		metadataStore.put("foo", "bar");
+		assertNull(metadataStore.putIfAbsent("foo", "baz"));
+		assertNotNull(metadataStore.putIfAbsent("foo", "baz"));
+		assertFalse(metadataStore.replace("foo", "xxx", "bar"));
+		assertTrue(metadataStore.replace("foo", "baz", "bar"));
 		metadataStore.destroy();
 		Properties persistentProperties = PropertiesLoaderUtils.loadProperties(new FileSystemResource(file));
 		assertNotNull(persistentProperties);

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2014 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,7 +15,7 @@
  */
 package org.springframework.integration.file.filters;
 
-import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.util.Assert;
 
 /**
@@ -30,13 +30,13 @@ import org.springframework.util.Assert;
  */
 public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
 
-	protected final MetadataStore store;
+	protected final ConcurrentMetadataStore store;
 
 	protected final String prefix;
 
 	private final Object monitor = new Object();
 
-	public AbstractPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+	public AbstractPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
 		Assert.notNull(store, "'store' cannot be null");
 		Assert.notNull(prefix, "'prefix' cannot be null");
 		this.store = store;
@@ -47,13 +47,16 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 	protected boolean accept(F file) {
 		String key = buildKey(file);
 		synchronized(monitor) {
-			String value = store.get(key);
-			if (value != null && isEqual(file, value)) {
+			String newValue = value(file);
+			String oldValue = this.store.putIfAbsent(key, newValue);
+			if (oldValue == null) { // not in store
+				return true;
+			}
+			if (isEqual(file, oldValue)) { // same value in store
 				return false;
 			}
-			store.put(key, value(file));
+			return this.store.replace(key, oldValue, newValue); // true if replace successful
 		}
-		return true;
 	}
 
 	/**
@@ -73,7 +76,7 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 	 * @return true if equal.
 	 */
 	protected boolean isEqual(F file, String value) {
-		return Long.valueOf(value).longValue() == this.modified(file);
+		return Long.valueOf(value) == this.modified(file);
 	}
 
 	/**

spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2014 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,7 +18,7 @@ package org.springframework.integration.file.filters;
 
 import java.io.File;
 
-import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 
 /**
  * @author Gary Russell
@@ -27,7 +27,7 @@ import org.springframework.integration.metadata.MetadataStore;
  */
 public class FileSystemPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<File> {
 
-	public FileSystemPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+	public FileSystemPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
 		super(store, prefix);
 	}
 

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterRedisTests.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2014 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.file.filters;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+import org.springframework.integration.redis.metadata.RedisMetadataStore;
+import org.springframework.integration.redis.rules.RedisAvailable;
+import org.springframework.integration.redis.rules.RedisAvailableTests;
+
+/**
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public class PersistentAcceptOnceFileListFilterRedisTests extends RedisAvailableTests {
+
+	@Before
+	@After
+	public void setupShutDown() {
+		RedisTemplate<String, ?> template = this.createTemplate();
+		template.delete("persistentAcceptOnceFileListFilterRedisTests");
+	}
+
+	private RedisTemplate<String, ?> createTemplate() {
+		RedisTemplate<String, ?> template = new RedisTemplate<String, Object>();
+		template.setConnectionFactory(this.getConnectionFactoryForTest());
+		template.setKeySerializer(new StringRedisSerializer());
+		template.afterPropertiesSet();
+		return template;
+	}
+
+	@Test
+	@RedisAvailable
+	public void testFileSystem() throws Exception {
+		final AtomicBoolean suspend = new AtomicBoolean();
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		RedisMetadataStore store = new RedisMetadataStore(this.getConnectionFactoryForTest(),
+				"persistentAcceptOnceFileListFilterRedisTests") {
+
+			@Override
+			public boolean replace(String key, String oldValue, String newValue) {
+				if (suspend.get()) {
+					latch2.countDown();
+					try {
+						latch1.await(10, TimeUnit.SECONDS);
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+					}
+				}
+				return super.replace(key, oldValue, newValue);
+			}
+
+		};
+		final FileSystemPersistentAcceptOnceFileListFilter filter =
+				new FileSystemPersistentAcceptOnceFileListFilter(store,"foo:");
+		final File file = File.createTempFile("foo", ".txt");
+		assertEquals(1, filter.filterFiles(new File[] {file}).size());
+		String ts = store.get("foo:" + file.getAbsolutePath());
+		assertEquals(String.valueOf(file.lastModified()), ts);
+		assertEquals(0, filter.filterFiles(new File[] {file}).size());
+		file.setLastModified(file.lastModified() + 5000L);
+		assertEquals(1, filter.filterFiles(new File[] {file}).size());
+		ts = store.get("foo:" + file.getAbsolutePath());
+		assertEquals(String.valueOf(file.lastModified()), ts);
+		assertEquals(0, filter.filterFiles(new File[] {file}).size());
+
+		suspend.set(true);
+		file.setLastModified(file.lastModified() + 5000L);
+
+		Future<Integer> result = Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
+
+			@Override
+			public Integer call() throws Exception {
+				return filter.filterFiles(new File[] {file}).size();
+			}
+		});
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		store.put("foo:" + file.getAbsolutePath(), "43");
+		latch1.countDown();
+		Integer theResult = result.get(10, TimeUnit.SECONDS);
+		assertEquals(Integer.valueOf(0), theResult); // lost the race, key changed
+
+		file.delete();
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2014 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,13 +15,20 @@
  */
 package org.springframework.integration.file.filters;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
 
-import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
@@ -33,14 +40,55 @@ public class PersistentAcceptOnceFileListFilterTests {
 
 	@Test
 	public void testFileSystem() throws Exception {
-		MetadataStore store = new SimpleMetadataStore();
-		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(store, "foo:");
-		File file = File.createTempFile("foo", ".txt");
-		assertTrue(filter.filterFiles(new File[] {file}).size() == 1);
-		assertTrue(filter.filterFiles(new File[] {file}).size() == 0);
-		file.setLastModified(27L);
-		assertTrue(filter.filterFiles(new File[] {file}).size() == 1);
-		assertTrue(filter.filterFiles(new File[] {file}).size() == 0);
+		final AtomicBoolean suspend = new AtomicBoolean();
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(1);
+		ConcurrentMetadataStore store = new SimpleMetadataStore() {
+
+			@Override
+			public boolean replace(String key, String oldValue, String newValue) {
+				if (suspend.get()) {
+					latch2.countDown();
+					try {
+						latch1.await(10, TimeUnit.SECONDS);
+					}
+					catch (InterruptedException e) {
+						Thread.currentThread().interrupt();
+					}
+				}
+				return super.replace(key, oldValue, newValue);
+			}
+
+		};
+		final FileSystemPersistentAcceptOnceFileListFilter filter =
+				new FileSystemPersistentAcceptOnceFileListFilter(store, "foo:");
+		final File file = File.createTempFile("foo", ".txt");
+		assertEquals(1, filter.filterFiles(new File[] {file}).size());
+		String ts = store.get("foo:" + file.getAbsolutePath());
+		assertEquals(String.valueOf(file.lastModified()), ts);
+		assertEquals(0, filter.filterFiles(new File[] {file}).size());
+		file.setLastModified(file.lastModified() + 5000L);
+		assertEquals(1, filter.filterFiles(new File[] {file}).size());
+		ts = store.get("foo:" + file.getAbsolutePath());
+		assertEquals(String.valueOf(file.lastModified()), ts);
+		assertEquals(0, filter.filterFiles(new File[] {file}).size());
+
+		suspend.set(true);
+		file.setLastModified(file.lastModified() + 5000L);
+
+		Future<Integer> result = Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
+
+			@Override
+			public Integer call() throws Exception {
+				return filter.filterFiles(new File[] {file}).size();
+			}
+		});
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		store.put("foo:" + file.getAbsolutePath(), "43");
+		latch1.countDown();
+		Integer theResult = result.get(10, TimeUnit.SECONDS);
+		assertEquals(Integer.valueOf(0), theResult); // lost the race, key changed
+
 		file.delete();
 	}
 

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2014 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.
@@ -19,12 +19,11 @@ package org.springframework.integration.ftp.filters;
 import org.apache.commons.net.ftp.FTPFile;
 
 import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
-import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 
 /**
- * Since the super class deems files as 'not seen' if the timestamp is different, remote file
- * users should use the adapter's preserve-timestamp option. Otherwise if a file is re-fetched
- * it will have a new timestamp.
+ * Persistent file list filter using the server's file timestamp to detect if we've already
+ * 'seen' this file.
  *
  * @author Gary Russell
  * @since 3.0
@@ -32,7 +31,7 @@ import org.springframework.integration.metadata.MetadataStore;
  */
 public class FtpPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<FTPFile> {
 
-	public FtpPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+	public FtpPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
 		super(store, prefix);
 	}
 

spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java

@@ -18,6 +18,7 @@ import org.springframework.data.redis.core.BoundHashOperations;
 import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.support.collections.RedisProperties;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.metadata.MetadataStore;
 import org.springframework.util.Assert;
 
@@ -29,7 +30,7 @@ import org.springframework.util.Assert;
  * @author Artem Bilan
  * @since 3.0
  */
-public class RedisMetadataStore implements MetadataStore {
+public class RedisMetadataStore implements ConcurrentMetadataStore {
 
 	public static final String KEY = "MetaData";
 
@@ -113,14 +114,41 @@ public class RedisMetadataStore implements MetadataStore {
 	@Override
 	public String get(String key) {
 		Assert.notNull(key, "'key' must not be null.");
-		return (String) this.properties.get(key);
+		Object value = this.properties.get(key);
+		if (value != null) {
+			Assert.isInstanceOf(String.class, value, "Invalid type in the store");
+		}
+		return (String) value;
 	}
 
 	@Override
 
 	public String remove(String key) {
 		Assert.notNull(key, "'key' must not be null.");
-		return (String) this.properties.remove(key);
+		Object removed = this.properties.remove(key);
+		if (removed != null) {
+			Assert.isInstanceOf(String.class, removed, "The removed value was an invalid type");
+		}
+		return (String) removed;
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(value, "'value' must not be null.");
+		Object oldValue = this.properties.putIfAbsent(key, value);
+		if (oldValue != null) {
+			Assert.isInstanceOf(String.class, oldValue, "Invalid type in the store");
+		}
+		return (String) oldValue;
+	}
+
+	@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.");
+		return this.properties.replace(key, oldValue, newValue);
 	}
 
 }

spring-integration-redis/src/test/java/org/springframework/integration/redis/util/RedisLockRegistryTests.java

@@ -57,7 +57,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
 
 	@Before
 	@After
-	public void shutDown() {
+	public void setupShutDown() {
 		RedisTemplate<String, ?> template = this.createTemplate();
 		template.delete("rlrTests");
 		template.delete("rlrTests2");

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2014 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,14 +18,13 @@ package org.springframework.integration.sftp.filters;
 
 
 import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
-import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 
 /**
- * Since the super class deems files as 'not seen' if the timestamp is different, remote file
- * users should use the adapter's preserve-timestamp option. Otherwise if a file is re-fetched
- * it will have a new timestamp.
+ * Persistent file list filter using the server's file timestamp to detect if we've already
+ * 'seen' this file.
  *
  * @author Gary Russell
  * @since 3.0
@@ -33,7 +32,7 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  */
 public class SftpPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<LsEntry> {
 
-	public SftpPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+	public SftpPersistentAcceptOnceFileListFilter(ConcurrentMetadataStore store, String prefix) {
 		super(store, prefix);
 	}
 

src/reference/docbook/aggregator.xml

@@ -535,6 +535,9 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
 			used to obtain a <interfacename>Lock</interfacename> based on the <code>groupId</code> for
 			concurrent operations on the
 			<code>MessageGroup</code>. By default, an internal <classname>DefaultLockRegistry</classname> is used.
+			Use of a distributed <interfacename>LockRegistry</interfacename>, such as the <classname
+			>RedisLockRegistry</classname>, ensures only one instance of the aggregator will operate
+			on a group concurrently. See <xref linkend="redis-lock-registry"/> for more information.
         </para>
       </callout>
 

src/reference/docbook/file.xml

@@ -36,11 +36,21 @@
       <classname>AcceptOnceFileListFilter</classname> is used. This filter
       ensures files are picked up only once from the directory.
       <note>
+       <para>
         The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
         state to survive a system restart, consider using the
         <classname>FileSystemPersistentAcceptOnceFileListFilter</classname> instead. This filter stores
-        the accepted file names in a <interfacename>MetadataStore</interfacename> strategy (<xref linkend="metadata-store"/>).
+        the accepted file names in a <interfacename>MetadataStore</interfacename> implementation
+        (<xref linkend="metadata-store"/>).
         This filter matches on the filename and modified time.
+       </para>
+       <para>
+        Since <emphasis>version 4.0</emphasis>, this filter requires a <interfacename
+        >ConcurrentMetadataStore</interfacename>. When used with a shared data store
+        (such as <code>Redis</code> with the <classname>RedisMetadataStore</classname>) this allows
+        filter keys to be shared across multiple application instances, or when a network
+        file share is being used by multiple servers.
+       </para>
       </note>
       <programlisting language="xml"><![CDATA[<bean id="pollableFileSource"
     class="org.springframework.integration.file.FileReadingMessageSource"

src/reference/docbook/ftp.xml

@@ -195,6 +195,12 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
         the accepted file names in an instance of the
         <interfacename>MetadataStore</interfacename> strategy (<xref linkend="metadata-store"/>).
         This filter matches on the filename and the remote modified time.
+    </para>
+    <para>
+        Since <emphasis>version 4.0</emphasis>, this filter requires a <interfacename
+        >ConcurrentMetadataStore</interfacename>. When used with a shared data store
+        (such as <code>Redis</code> with the <classname>RedisMetadataStore</classname>) this allows
+        filter keys to be shared across multiple application or server instances.
     </para>
 	<note>
 	  <para>

src/reference/docbook/meta-data-store.xml

@@ -40,6 +40,12 @@
 		<interfacename>MetadataStore</interfacename> interface (e.g. JdbcMetadataStore)
 		and configure it as a bean in the Application Context.
 	</para>
+	<para>
+		Starting with <emphasis>version 4.0</emphasis>, <classname>SimpleMetadataStore</classname>,
+		<classname>PropertiesPersistingMetadataStore</classname> and
+		<classname>RedisMetadataStore</classname> implement <interfacename>ConcurrentMetadataStore</interfacename>.
+		These provide for atomic updates and can be used across multiple component or application instances.
+	</para>
 	<section id="idempotent-receiver">
 		<title>Idempotent Receiver</title>
 		<para>

src/reference/docbook/redis.xml

@@ -489,6 +489,11 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
 			<code>key</code> plays the role of a <emphasis>region</emphasis>, which is useful in distributed environment,
 			when several applications use the same Redis server. By default this <code>key</code> has the value <code>MetaData</code>.
 		</para>
+		<para>
+			Starting with <emphasis>version 4.0</emphasis>, this store now implements <interfacename
+			>ConcurrentMetadataStore</interfacename>, allowing it to be reliably shared across multiple
+			application instances where only one instance will be allowed to store or modify a key's value.
+		</para>
 	</section>
   <section id="redis-store-inbound-channel-adapter">
         <title>RedisStore Inbound Channel Adapter</title>
@@ -770,4 +775,34 @@ the serialization of values, you may want to consider providing your own
 		  <ulink url="http://redis.io/commands">Redis Specification</ulink>.
 		</para>
 	</section>
+	<section id="redis-lock-registry">
+		<title>Redis Lock Registry</title>
+		<para>
+			Starting with <emphasis>version 4.0</emphasis>, the <classname>RedisLockRegistry</classname> is
+			available. Certain components (for example aggregator and resequencer) use a lock obtained from
+			a <interfacename>LockRegistry</interfacename> instance to ensure
+			that only one thread is manipulating a group at a time. The <classname>DefaultLockRegistry</classname>
+			performs this function within a single component; you can now configure an external lock registry
+			on these components. When used with a shared <interfacename>MessageGroupStore</interfacename>,
+			the <classname>RedisLockRegistry</classname> can be use to provide this functionality across
+			multiple application instances, such that only one instance can manipulate the group at a time.
+		</para>
+		<para>
+			When a lock is released by a local thread, another local thread will generally be able to acquire the
+			lock immediately. If a lock is released by a thread using a different registry instance, it can take up to
+			100ms to acquire the lock.
+		</para>
+		<para>
+			To avoid "hung" locks (when a server fails), the locks in this registry are expired after a default
+			60 seconds, but this can be configured on the registry. Locks are normally held for a much smaller
+			time. 
+		</para>
+		<important>
+			Because the keys can expire, an attempt to unlock an expired lock will result in an exception
+			being thrown. However, be aware that the resources protected by such a lock may have been
+			compromised so such exceptions should be considered severe. The expiry should be set at
+			a large enough value to prevent this condition, while small enough that the lock can
+			be recovered after a server failure in a reasonable amount of time. 
+		</important>
+	</section>
 </chapter>

src/reference/docbook/resequencer.xml

@@ -56,9 +56,9 @@
 
   lock-registry="lockRegistry" ]]><co id="resxml18" /><![CDATA[
 
-		group-timeout="60000" ]]><co id="resxml19" /><![CDATA[
-		group-timeout-expression="size() ge 2 ? 100 : -1" ]]><co id="resxml20" /><![CDATA[
-		scheduler="taskScheduler" /> ]]><co id="resxml21" /></programlisting>
+  group-timeout="60000" ]]><co id="resxml19" /><![CDATA[
+  group-timeout-expression="size() ge 2 ? 100 : -1" ]]><co id="resxml20" /><![CDATA[
+  scheduler="taskScheduler" /> ]]><co id="resxml21" /></programlisting>
     <para><calloutlist>
         <callout arearefs="resxml1-co" id="resxml1">
           <para>The id of the resequencer is

src/reference/docbook/sftp.xml

@@ -320,6 +320,12 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
         the accepted file names in an instance of the
         <interfacename>MetadataStore</interfacename> strategy (<xref linkend="metadata-store"/>).
         This filter matches on the filename and the remote modified time.
+    </para>
+    <para>
+        Since <emphasis>version 4.0</emphasis>, this filter requires a <interfacename
+        >ConcurrentMetadataStore</interfacename>. When used with a shared data store
+        (such as <code>Redis</code> with the <classname>RedisMetadataStore</classname>) this allows
+        filter keys to be shared across multiple application or server instances.
     </para>
 	<note>
 	  <para>

src/reference/docbook/whats-new.xml

@@ -133,6 +133,15 @@
 				For more information, see <xref linkend="redis-outbound-gateway"/>.
 			</para>
 		</section>
+		<section id="4.0-redis-lock-registry">
+			<title>RedisLockRegistry</title>
+			<para>
+				The <classname>RedisLockRegistry</classname> is now available supporting global locks visible
+				to multiple application instances/servers. This can be used with aggregating message handlers
+				across multiple application instances such that group release will occur on only one instance.
+				For more information, see <xref linkend="redis-lock-registry"/> and <xref linkend="aggregator"/>.
+			</para>
+		</section>
 	</section>
 
 	<section id="4.0-general">
@@ -220,5 +229,17 @@
 				For more information see <xref linkend="aggregator-config"/>.
 			</para>
 		</section>
+		<section id="4.0-redis-metadata">
+			<title>Redis Metadata Store</title>
+			<para>
+				The <classname>RedisMetadataStore</classname> now implements <interfacename
+				>ConcurrentMetadataStore</interfacename>, allowing it to be used, for example,
+				in a <classname>AbstractPersistentAcceptOnceFileListFilter</classname>
+				implementation in a multiple application instance/server environment.
+				For more information, see <xref linkend="redis-metadata-store"/>,  
+				<xref linkend="file-reading"/>, <xref linkend="ftp-inbound"/> and
+				<xref linkend="sftp-inbound"/>.
+			</para>
+		</section>
 	</section>
 </chapter>