INT-4123: Add Prefix to the Key-Value MSs

Fixes spring-projects/spring-integration#2213
JIRA: https://jira.spring.io/browse/INT-4123

Fully different `MessageStore`s can be configured for the same shared
Key-Value data-base.
Since the retrieval logic is based on the keys, that may cause the
unexpected messages expiration via `MessageGroupStoreReaper`.

* To distinguish store instances on the shared store add `prefix`
option to the `AbstractKeyValueMessageStore`

* Deprecate the `GemfireMessageStore` `Cache`-based configuration - `setIgnoreJta()` and `afterPropertiesSet()`.
The `GemfireMessageStore` relies only on an externally configured `Region`.

**Cherry-pick to 4.3.x**

Doc Polishing

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/store/AbstractKeyValueMessageStore.java
#	spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java
#	spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisMessageStore.java

* Fix tests `GemfireMessageStore` tests conflicts

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java

@@ -39,16 +39,18 @@ import com.gemstone.gemfire.cache.Region;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author David Turanski
+ * @author Artem Bilan
+ *
  * @since 2.1
  */
 public class GemfireMessageStore extends AbstractKeyValueMessageStore implements InitializingBean {
 
 	private static final String MESSAGE_STORE_REGION_NAME = "messageStoreRegion";
 
-	private volatile Region<Object, Object> messageStoreRegion;
-
 	private final Cache cache;
 
+	private volatile Region<Object, Object> messageStoreRegion;
+
 	private volatile boolean ignoreJta = true;
 
 	/**
@@ -58,6 +60,19 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements
 	 * @param messageStoreRegion The region.
 	 */
 	public GemfireMessageStore(Region<Object, Object> messageStoreRegion) {
+		this(messageStoreRegion, "");
+	}
+
+	/**
+	 * Construct a {@link GemfireMessageStore} instance based on the provided
+	 * @param messageStoreRegion the region to use.
+	 * @param prefix the key prefix to use, allowing the same region to be used for
+	 * multiple stores.
+	 * @since 4.3.12
+	 */
+	public GemfireMessageStore(Region<Object, Object> messageStoreRegion, String prefix) {
+		super(prefix);
+		Assert.notNull(messageStoreRegion, "'messageStoreRegion' must not be null");
 		this.cache = null;
 		this.messageStoreRegion = messageStoreRegion;
 	}
@@ -66,7 +81,6 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements
 	 * Provides a cache reference used to create a message store region named
 	 * 'messageStoreRegion'
 	 * @param cache The cache.
-	 *
 	 * @deprecated - use the other constructor and provide a region directly.
 	 */
 	@Deprecated
@@ -75,12 +89,19 @@ public class GemfireMessageStore extends AbstractKeyValueMessageStore implements
 		this.cache = cache;
 	}
 
+	/**
+	 * The boolean flag to ignore JTA on the Gemfire Region.
+	 * @param ignoreJta boolean flag to ignore JTA on the Gemfire Region.
+	 * @deprecated with no-op, in favor of externally configured region.
+	 */
+	@Deprecated
 	public void setIgnoreJta(boolean ignoreJta) {
 		this.ignoreJta = ignoreJta;
 	}
 
 	@Override
 	@SuppressWarnings({ "unchecked", "deprecation" })
+	@Deprecated
 	public void afterPropertiesSet() {
 		if (this.messageStoreRegion != null) {
 			return;

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests-context.xml

@@ -7,7 +7,7 @@
 
 	<beans:bean id="messageStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
 		<beans:constructor-arg
-				value="#{T (org.springframework.integration.gemfire.store.DelayerHandlerRescheduleIntegrationTests).cacheFactoryBean.object}"/>
+				value="#{T (org.springframework.integration.gemfire.store.DelayerHandlerRescheduleIntegrationTests).region}"/>
 	</beans:bean>
 
 	<channel id="output">

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java

@@ -44,6 +44,9 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.Scope;
 
 /**
  * @author Artem Bilan
@@ -55,6 +58,8 @@ public class DelayerHandlerRescheduleIntegrationTests {
 
 	public static final String DELAYER_ID = "delayerWithGemfireMS";
 
+	public static Region<Object, Object> region;
+
 	public static CacheFactoryBean cacheFactoryBean;
 
 	@ClassRule
@@ -67,10 +72,15 @@ public class DelayerHandlerRescheduleIntegrationTests {
 		gemfireProperties.setProperty("mcast-port", "0");
 		cacheFactoryBean.setProperties(gemfireProperties);
 		cacheFactoryBean.afterPropertiesSet();
+		Cache cache = cacheFactoryBean.getObject();
+		region = cache.createRegionFactory().setScope(Scope.LOCAL).create("sig-tests");
 	}
 
 	@AfterClass
 	public static void cleanUp() throws Exception {
+		if (region != null) {
+			region.close();
+		}
 		if (cacheFactoryBean != null) {
 			cacheFactoryBean.destroy();
 		}
@@ -95,7 +105,7 @@ public class DelayerHandlerRescheduleIntegrationTests {
 				(ThreadPoolTaskScheduler) IntegrationContextUtils.getTaskScheduler(context);
 		taskScheduler.shutdown();
 		taskScheduler.getScheduledExecutor().awaitTermination(10, TimeUnit.SECONDS);
-		context.destroy();
+		context.close();
 
 		try {
 			context.getBean("input", MessageChannel.class);
@@ -135,11 +145,11 @@ public class DelayerHandlerRescheduleIntegrationTests {
 		assertEquals(1, messageStore.getMessageGroupCount());
 		int n = 0;
 		while (n++ < 200 && messageStore.messageGroupSize(delayerMessageGroupId) > 0) {
-			Thread.sleep(50);
+			Thread.sleep(100);
 		}
 		assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId));
 
-		context.destroy();
+		context.close();
 	}
 
 }

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java

@@ -64,12 +64,11 @@ public class GemfireGroupStoreTests {
 
 	public static CacheFactoryBean cacheFactoryBean;
 
-	private static Region<Object, Object> region;
+	public static Region<Object, Object> region;
 
 	@Test
 	public void testNonExistingEmptyMessageGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		assertNotNull(messageGroup);
 		assertTrue(messageGroup instanceof SimpleMessageGroup);
@@ -79,7 +78,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testMessageGroupWithAddedMessage() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> message = new GenericMessage<String>("Hello");
 		messageGroup = store.addMessageToGroup(1, message);
@@ -87,7 +85,6 @@ public class GemfireGroupStoreTests {
 
 		// make sure the store is properly rebuild from Gemfire
 		store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 
 		messageGroup = store.getMessageGroup(1);
 		assertEquals(1, messageGroup.size());
@@ -96,7 +93,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testRemoveMessageFromTheGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> message = new GenericMessage<String>("2");
 
@@ -121,7 +117,6 @@ public class GemfireGroupStoreTests {
 
 		// make sure the store is properly rebuild from Gemfire
 		store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 
 		messageGroup = store.getMessageGroup(1);
 		assertEquals(2, messageGroup.size());
@@ -131,7 +126,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testRemoveMessageGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> message = new GenericMessage<String>("Hello");
 		messageGroup = store.addMessageToGroup(messageGroup.getGroupId(), message);
@@ -145,7 +139,6 @@ public class GemfireGroupStoreTests {
 
 		// make sure the store is properly rebuild from Gemfire
 		store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 
 		messageGroup = store.getMessageGroup(1);
 
@@ -156,7 +149,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testRemoveNonExistingMessageFromTheGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		store.addMessagesToGroup(messageGroup.getGroupId(), new GenericMessage<String>("1"));
 		store.removeMessagesFromGroup(1, new GenericMessage<String>("2"));
@@ -165,14 +157,12 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testRemoveNonExistingMessageFromNonExistingTheGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		store.removeMessagesFromGroup(1, new GenericMessage<String>("2"));
 	}
 
 	@Test
 	public void testCompleteMessageGroup() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> messageToMark = new GenericMessage<String>("1");
 		store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark);
@@ -184,7 +174,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testLastReleasedSequenceNumber() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 		MessageGroup messageGroup = store.getMessageGroup(1);
 		Message<?> messageToMark = new GenericMessage<String>("1");
 		store.addMessagesToGroup(messageGroup.getGroupId(), messageToMark);
@@ -196,10 +185,8 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testMultipleInstancesOfGroupStore() throws Exception {
 		GemfireMessageStore store1 = new GemfireMessageStore(region);
-		store1.afterPropertiesSet();
 
 		GemfireMessageStore store2 = new GemfireMessageStore(region);
-		store2.afterPropertiesSet();
 
 		Message<?> message = new GenericMessage<String>("1");
 		store1.addMessagesToGroup(1, message);
@@ -208,7 +195,6 @@ public class GemfireGroupStoreTests {
 		assertEquals(2, messageGroup.getMessages().size());
 
 		GemfireMessageStore store3 = new GemfireMessageStore(region);
-		store3.afterPropertiesSet();
 
 		store3.removeMessagesFromGroup(1, message);
 		messageGroup = store3.getMessageGroup(1);
@@ -219,7 +205,6 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testWithMessageHistory() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 
 		store.getMessageGroup(1);
 
@@ -246,9 +231,7 @@ public class GemfireGroupStoreTests {
 	@Test
 	public void testIteratorOfMessageGroups() throws Exception {
 		GemfireMessageStore store1 = new GemfireMessageStore(region);
-		store1.afterPropertiesSet();
 		GemfireMessageStore store2 = new GemfireMessageStore(region);
-		store2.afterPropertiesSet();
 
 		store1.addMessagesToGroup(1, new GenericMessage<String>("1"));
 		store2.addMessagesToGroup(2, new GenericMessage<String>("2"));
@@ -278,9 +261,7 @@ public class GemfireGroupStoreTests {
 	public void testConcurrentModifications() throws Exception {
 
 		final GemfireMessageStore store1 = new GemfireMessageStore(region);
-		store1.afterPropertiesSet();
 		final GemfireMessageStore store2 = new GemfireMessageStore(region);
-		store2.afterPropertiesSet();
 
 		final Message<?> message = new GenericMessage<String>("1");
 
@@ -345,7 +326,7 @@ public class GemfireGroupStoreTests {
 		Message<?> m3 = MessageBuilder.withPayload("3").setSequenceNumber(3).setSequenceSize(3).setCorrelationId(1)
 				.build();
 		inputA.send(m3);
-		assertNotNull(outputA.receive(1000));
+		assertNotNull(outputA.receive(10000));
 		context1.close();
 		context2.close();
 	}

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java

@@ -34,6 +34,7 @@ import org.springframework.data.gemfire.RegionFactoryBean;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupMetadata;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
@@ -60,8 +61,6 @@ public class GemfireMessageStoreTests {
 	@Test
 	public void addAndGetMessage() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
-
 		Message<?> message = MessageBuilder.withPayload("test").build();
 		store.addMessage(message);
 		Message<?> retrieved = store.getMessage(message.getHeaders().getId());
@@ -76,7 +75,6 @@ public class GemfireMessageStoreTests {
 		region.afterPropertiesSet();
 
 		GemfireMessageStore store = new GemfireMessageStore(region.getObject());
-		store.afterPropertiesSet();
 		assertSame(region.getObject(), TestUtils.getPropertyValue(store, "messageStoreRegion"));
 
 		region.destroy();
@@ -85,7 +83,6 @@ public class GemfireMessageStoreTests {
 	@Test
 	public void testWithMessageHistory() throws Exception {
 		GemfireMessageStore store = new GemfireMessageStore(region);
-		store.afterPropertiesSet();
 
 		Message<?> message = new GenericMessage<String>("Hello");
 		DirectChannel fooChannel = new DirectChannel();
@@ -108,7 +105,6 @@ public class GemfireMessageStoreTests {
 	@Test
 	public void testAddAndRemoveMessagesFromMessageGroup() throws Exception {
 		GemfireMessageStore messageStore = new GemfireMessageStore(region);
-		messageStore.afterPropertiesSet();
 
 		String groupId = "X";
 		List<Message<?>> messages = new ArrayList<Message<?>>();
@@ -124,6 +120,29 @@ public class GemfireMessageStoreTests {
 		assertEquals(0, group.size());
 	}
 
+	@Test
+	public void testAddAndRemoveMessagesFromMessageGroupWithPrefix() throws Exception {
+		GemfireMessageStore messageStore = new GemfireMessageStore(region, "foo_");
+
+		String groupId = "X";
+		List<Message<?>> messages = new ArrayList<Message<?>>();
+		for (int i = 0; i < 25; i++) {
+			Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
+			messageStore.addMessagesToGroup(groupId, message);
+			messages.add(message);
+		}
+
+		MessageGroupMetadata messageGroupMetadata =
+				(MessageGroupMetadata) region.get("foo_" + "MESSAGE_GROUP_" + groupId);
+
+		assertNotNull(messageGroupMetadata);
+		assertEquals(25, messageGroupMetadata.size());
+
+		messageStore.removeMessagesFromGroup(groupId, messages);
+		MessageGroup group = messageStore.getMessageGroup(groupId);
+		assertEquals(0, group.size());
+	}
+
 	@Before
 	public void prepare() {
 		if (region != null) {

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config-a.xml

@@ -12,8 +12,7 @@
 	</int:channel>
 
 	<bean id="gemfireStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
-		<constructor-arg
-				value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).cacheFactoryBean.object}"/>
+		<constructor-arg value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).region}"/>
 	</bean>
 
 </beans>

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-aggregator-config.xml

@@ -12,8 +12,7 @@
 	</int:channel>
 
 	<bean id="gemfireStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
-		<constructor-arg
-				value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).cacheFactoryBean.object}"/>
+		<constructor-arg value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).region}"/>
 	</bean>
 
 </beans>

spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/gemfire-queue-config.xml

@@ -19,8 +19,7 @@
 	</int:channel>
 
 	<bean id="gemfireStore" class="org.springframework.integration.gemfire.store.GemfireMessageStore">
-		<constructor-arg
-				value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).cacheFactoryBean.object}"/>
+		<constructor-arg value="#{T (org.springframework.integration.gemfire.store.GemfireGroupStoreTests).region}"/>
 	</bean>
 
 </beans>