INT-3704 Add Map ctor for SimpleMetadataStore

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

Some NoSQLs (e.g. Hazelcast) provide a distributed implementations for the `ConcurrentMap`,
so add `SimpleMetadataStore(ConcurrentMap<String, String> metadata)` to allow ot inject any `ConcurrentMap` implementation.
Demonstrate the support for Hazelcast in the `IdempotentReceiverIntegrationTests`

**Cherry-pick to 4.1.x**

build.gradle

@@ -92,6 +92,7 @@ subprojects { subproject ->
 		groovyVersion = '2.4.0'
 		guavaVersion = '18.0'
 		hamcrestVersion = '1.3'
+		hazelcastVersion = '3.4.2'
 		hibernateVersion = '4.3.8.Final'
 		hsqldbVersion = '2.3.2'
 		h2Version = '1.4.180'
@@ -399,6 +400,7 @@ project('spring-integration-jmx') {
 	dependencies {
 		compile project(":spring-integration-core")
 		testCompile "org.aspectj:aspectjweaver:$aspectjVersion"
+		testCompile "com.hazelcast:hazelcast:$hazelcastVersion"
 	}
 }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * 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
@@ -16,19 +16,40 @@ package org.springframework.integration.metadata;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import org.springframework.util.Assert;
+
 
 /**
- * Simple implementation of {@link MetadataStore} that uses an in-memory map only.
- * The metadata will not be persisted across application restarts.
+ * Simple implementation of {@link MetadataStore} that uses a {@link ConcurrentMap} for the data store.
+ * The metadata may not be persisted across application restarts, if the provided {@link ConcurrentMap}
+ * is an in-memory instance.
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  */
 public class SimpleMetadataStore implements ConcurrentMetadataStore {
 
-	private final ConcurrentMap<String, String> metadata = new ConcurrentHashMap<String, String>();
+	private final ConcurrentMap<String, String> metadata;
 
+	/**
+	 * Instantiate a {@link SimpleMetadataStore} using an in-memory {@link ConcurrentHashMap}.
+	 */
+	public SimpleMetadataStore() {
+		this(new ConcurrentHashMap<String, String>());
+	}
+
+	/**
+	 * Instantiate a {@link SimpleMetadataStore} using the provided {@link ConcurrentMap}.
+	 * The implementation may be a distributed map provided by projects such as Redis and Hazelcast.
+	 * @param metadata the {@link ConcurrentMap} instance for metadata.
+	 * @since 4.1.4
+	 */
+	public SimpleMetadataStore(ConcurrentMap<String, String> metadata) {
+		Assert.notNull(metadata, "'metadata' must not be null.");
+		this.metadata = metadata;
+	}
 
 	@Override
 	public void put(String key, String value) {

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java

@@ -69,6 +69,10 @@ import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import com.hazelcast.config.Config;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+
 /**
  * @author Artem Bilan
  * @since 4.1
@@ -183,9 +187,15 @@ public class IdempotentReceiverIntegrationTests {
 			return new MBeanServerFactoryBean();
 		}
 
+		@Bean
+		public HazelcastInstance hazelcastInstance() {
+			return Hazelcast.newHazelcastInstance(new Config().setProperty( "hazelcast.logging.type", "log4j" ));
+		}
+
+
 		@Bean
 		public ConcurrentMetadataStore store() {
-			return new SimpleMetadataStore();
+			return new SimpleMetadataStore(hazelcastInstance().getMap("idempotentReceiverMetadataStore"));
 		}
 
 		@Bean
@@ -296,7 +306,7 @@ public class IdempotentReceiverIntegrationTests {
 	@Component
 	private static class FooService {
 
-		private List<Message<?>> messages = new ArrayList<Message<?>>();
+		private final List<Message<?>> messages = new ArrayList<Message<?>>();
 
 		@ServiceActivator(inputChannel = "annotatedMethodChannel")
 		@IdempotentReceiver("idempotentReceiverInterceptor")