INT-2912 JdbcMessageStore Add 'region' to Queries

Missing region column in some queries.

For reference see: https://jira.springsource.org/browse/INT-2912

INT-2912 - Code Review
* Eliminate Spring Application Context for Test
* Improve SQL Query

INT-2912 Polishing

Modified testVerifyMessageGroupCount to test size of each
group in each region.

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2013 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
@@ -89,7 +89,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	public static final String DEFAULT_TABLE_PREFIX = "INT_";
 
 	private enum Query {
-		GROUP_EXISTS("SELECT COUNT(GROUP_KEY) FROM %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"),
+		GROUP_EXISTS("SELECT COUNT(GROUP_KEY) FROM %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"),
 
 		CREATE_MESSAGE_GROUP("INSERT into %PREFIX%MESSAGE_GROUP" +
 			"(GROUP_KEY, REGION, MARKED, COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE)"
@@ -97,16 +97,16 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 		UPDATE_MESSAGE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"),
 
-		REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"),
+		REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=? and REGION=?"),
 
-		REMOVE_GROUP_TO_MESSAGE_JOIN("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
+		REMOVE_GROUP_TO_MESSAGE_JOIN("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and REGION=?"),
 
-		COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"),
+		COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where REGION=?"),
 
-		COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"),
+		COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and REGION=?"),
 
 		LIST_MESSAGEIDS_BY_GROUP_KEY("select MESSAGE_ID, CREATED_DATE " +
-				"from %PREFIX%MESSAGE where MESSAGE_ID in (select MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY = ?) and REGION=? " +
+				"from %PREFIX%MESSAGE where MESSAGE_ID in (select MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and REGION=?) " +
 				"ORDER BY CREATED_DATE"),
 
 		LIST_MESSAGES_BY_GROUP_KEY("SELECT MESSAGE_ID, MESSAGE_BYTES, CREATED_DATE " +
@@ -122,7 +122,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 						"and %PREFIX%MESSAGE.REGION = ?))"),
 
 		GET_GROUP_INFO("SELECT COMPLETE, LAST_RELEASED_SEQUENCE, CREATED_DATE, UPDATED_DATE" +
-				" from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ?"),
+				" from %PREFIX%MESSAGE_GROUP where GROUP_KEY = ? and REGION=?"),
 
 		GET_MESSAGE("SELECT MESSAGE_ID, CREATED_DATE, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=? and REGION=?"),
 
@@ -144,8 +144,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 		DELETE_MESSAGE_GROUP("DELETE from %PREFIX%MESSAGE_GROUP where GROUP_KEY=? and REGION=?"),
 
 		CREATE_GROUP_TO_MESSAGE("INSERT into %PREFIX%GROUP_TO_MESSAGE" +
-				"(GROUP_KEY, MESSAGE_ID)"
-				+ " values (?, ?)"),
+				"(GROUP_KEY, MESSAGE_ID, REGION)"
+				+ " values (?, ?, ?)"),
 
 		UPDATE_GROUP("UPDATE %PREFIX%MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?"),
 
@@ -226,6 +226,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	 * @param region the region name to set
 	 */
 	public void setRegion(String region) {
+		Assert.hasText(region, "Region must not be null or empty.");
 		this.region = region;
 	}
 
@@ -353,7 +354,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
 		final String groupKey = getKey(groupId);
 		final String messageId = getKey(message.getHeaders().getId());
-		boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(Query.GROUP_EXISTS), groupKey) < 1;
+		boolean groupNotExist = jdbcTemplate.queryForInt(this.getQuery(Query.GROUP_EXISTS), groupKey, region) < 1;
 
 		final Timestamp updatedDate = new Timestamp(System.currentTimeMillis());
 
@@ -382,6 +383,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 				}
 				ps.setString(1, groupKey);
 				ps.setString(2, messageId);
+				ps.setString(3, region);
 			}
 		});
 		return getMessageGroup(groupId);
@@ -397,13 +399,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 	@Override
 	@ManagedAttribute
 	public int getMessageCountForAllMessageGroups() {
-		return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUPS));
+		return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUPS), region);
 	}
 
 	@ManagedAttribute
 	public int messageGroupSize(Object groupId) {
 		String key = getKey(groupId);
-		return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUP), key);
+		return jdbcTemplate.queryForInt(getQuery(Query.COUNT_ALL_MESSAGES_IN_GROUP), key, region);
 	}
 
 	public MessageGroup getMessageGroup(Object groupId) {
@@ -419,7 +421,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 			return new SimpleMessageGroup(groupId);
 		}
 
-		jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key},
+		jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key, region},
 				new RowCallbackHandler() {
 					public void processRow(ResultSet rs) throws SQLException {
 						updateDate.set(rs.getTimestamp("UPDATED_DATE"));
@@ -464,6 +466,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 				}
 				ps.setString(1, groupKey);
 				ps.setString(2, messageId);
+				ps.setString(3, region);
 			}
 		});
 		this.removeMessage(messageToRemove.getHeaders().getId());
@@ -485,6 +488,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 					logger.debug("Removing relationships for the group with group key=" + groupKey);
 				}
 				ps.setString(1, groupKey);
+				ps.setString(2, region);
 			}
 		});
 

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-db2.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES BLOB
+	MESSAGE_BYTES BLOB,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES BLOB
+	MESSAGE_BYTES BLOB,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-h2.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES LONGVARBINARY
+	MESSAGE_BYTES LONGVARBINARY,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-hsqldb.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES LONGVARBINARY
+	MESSAGE_BYTES LONGVARBINARY,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-mysql.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE DATETIME NOT NULL,
-	MESSAGE_BYTES BLOB
+	MESSAGE_BYTES BLOB,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 ) ENGINE=InnoDB;
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 ) ENGINE=InnoDB;
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE DATETIME NOT NULL,
-	UPDATED_DATE DATETIME DEFAULT NULL
+	UPDATED_DATE DATETIME DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 ) ENGINE=InnoDB;

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-oracle10g.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR2(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES BLOB
+	MESSAGE_BYTES BLOB,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR2(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR2(100),
 	MARKED NUMBER(19,0),
 	COMPLETE NUMBER(19,0),
 	LAST_RELEASED_SEQUENCE NUMBER(19,0),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-postgresql.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE TIMESTAMP NOT NULL,
-	MESSAGE_BYTES BYTEA
+	MESSAGE_BYTES BYTEA,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE TIMESTAMP NOT NULL,
-	UPDATED_DATE TIMESTAMP DEFAULT NULL
+	UPDATED_DATE TIMESTAMP DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sqlserver.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE DATETIME NOT NULL,
-	MESSAGE_BYTES IMAGE
+	MESSAGE_BYTES IMAGE,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 );
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 );
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE DATETIME NOT NULL,
-	UPDATED_DATE DATETIME DEFAULT NULL
+	UPDATED_DATE DATETIME DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 );

spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/schema-sybase.sql

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION VARCHAR(100),
 	CREATED_DATE DATETIME NOT NULL,
-	MESSAGE_BYTES IMAGE
+	MESSAGE_BYTES IMAGE,
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 ) LOCK DATAROWS;
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION VARCHAR(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 ) LOCK DATAROWS;
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION VARCHAR(100),
 	MARKED BIGINT,
 	COMPLETE BIGINT,
 	LAST_RELEASED_SEQUENCE BIGINT,
 	CREATED_DATE DATETIME NOT NULL,
-	UPDATED_DATE DATETIME DEFAULT NULL
+	UPDATED_DATE DATETIME DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 ) LOCK DATAROWS;

spring-integration-jdbc/src/main/sql/schema.sql.vpp

@@ -1,26 +1,29 @@
 -- Autogenerated: do not edit this file
 
 CREATE TABLE INT_MESSAGE  (
-	MESSAGE_ID CHAR(36) NOT NULL PRIMARY KEY,
+	MESSAGE_ID CHAR(36),
 	REGION ${VARCHAR}(100),
 	CREATED_DATE ${TIMESTAMP} NOT NULL,
-	MESSAGE_BYTES ${BLOB}
+	MESSAGE_BYTES ${BLOB},
+	constraint MESSAGE_PK primary key (MESSAGE_ID, REGION)
 )#if(${VOODOO}) ${VOODOO}#end;
 
 CREATE INDEX INT_MESSAGE_IX1 ON INT_MESSAGE (CREATED_DATE);
 
 CREATE TABLE INT_GROUP_TO_MESSAGE  (
-	GROUP_KEY CHAR(36) NOT NULL,
-	MESSAGE_ID CHAR(36) NOT NULL,
-	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, MESSAGE_ID)
+	GROUP_KEY CHAR(36),
+	MESSAGE_ID CHAR(36),
+	REGION ${VARCHAR}(100),
+	constraint GROUP_TO_MESSAG_PK primary key (GROUP_KEY, MESSAGE_ID, REGION)
 )#if(${VOODOO}) ${VOODOO}#end;
 
 CREATE TABLE INT_MESSAGE_GROUP  (
-    GROUP_KEY CHAR(36) NOT NULL PRIMARY KEY,
+	GROUP_KEY CHAR(36),
 	REGION ${VARCHAR}(100),
 	MARKED ${BIGINT},
 	COMPLETE ${BIGINT},
 	LAST_RELEASED_SEQUENCE ${BIGINT},
 	CREATED_DATE ${TIMESTAMP} NOT NULL,
-	UPDATED_DATE ${TIMESTAMP} DEFAULT NULL
+	UPDATED_DATE ${TIMESTAMP} DEFAULT NULL,
+	constraint MESSAGE_GROUP_PK primary key (GROUP_KEY, REGION)
 )#if(${VOODOO}) ${VOODOO}#end;

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreRegionTests.java

@@ -0,0 +1,172 @@
+/*
+ * Copyright 2002-2013 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.jdbc;
+
+import static org.junit.Assert.assertEquals;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+/**
+ * @author Gunnar Hillert
+ */
+public class JdbcMessageStoreRegionTests {
+
+	private static EmbeddedDatabase dataSource;
+	private JdbcTemplate jdbcTemplate;
+
+	private JdbcMessageStore messageStore1;
+	private JdbcMessageStore messageStore2;
+
+	@BeforeClass
+	public static void setupDatabase() {
+		dataSource = new EmbeddedDatabaseBuilder()
+				.setType(EmbeddedDatabaseType.H2)
+				.addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql")
+				.build();
+	}
+
+	@AfterClass
+	public static void shutDownDatabase() {
+		dataSource.shutdown();
+	}
+
+	@Before
+	public void beforeTest() {
+		this.jdbcTemplate = new JdbcTemplate(dataSource);
+		this.messageStore1 = new JdbcMessageStore(dataSource);
+		messageStore1.setRegion("region1");
+
+		this.messageStore2 = new JdbcMessageStore(dataSource);
+		this.messageStore2.setRegion("region2");
+	}
+
+	@After
+	public void afterTest() {
+		this.jdbcTemplate.execute("delete from INT_GROUP_TO_MESSAGE");
+		this.jdbcTemplate.execute("delete from INT_MESSAGE");
+		this.jdbcTemplate.execute("delete from INT_MESSAGE_GROUP");
+	}
+
+	@Test
+	public void testVerifyMessageCount() throws Exception {
+
+		messageStore1.addMessage(MessageBuilder.withPayload("payload1").build());
+		messageStore1.addMessage(MessageBuilder.withPayload("payload2").build());
+
+		messageStore2.addMessage(MessageBuilder.withPayload("payload1").build());
+		messageStore2.addMessage(MessageBuilder.withPayload("payload2").build());
+
+		assertEquals(2, messageStore1.getMessageCount());
+		assertEquals(2, messageStore2.getMessageCount());
+
+	}
+
+	@Test
+	public void testInsertNullRegion() throws Exception {
+
+		try {
+			messageStore1.setRegion(null);
+		}
+		catch (IllegalArgumentException e) {
+			Assert.assertEquals("Region must not be null or empty.", e.getMessage());
+			return;
+		}
+
+		Assert.fail("Expected an IllegalArgumentException to be thrown.");
+	}
+
+	@Test
+	public void testVerifyMessageGroupCount() throws Exception {
+
+		messageStore1.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+		messageStore1.addMessageToGroup("group2", MessageBuilder.withPayload("payload2").build());
+
+		messageStore2.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+		messageStore2.addMessageToGroup("group2", MessageBuilder.withPayload("payload2").build());
+
+		assertEquals(1, messageStore1.getMessageGroup("group1").getMessages().size());
+		assertEquals(1, messageStore2.getMessageGroup("group1").getMessages().size());
+		assertEquals(1, messageStore1.getMessageGroup("group2").getMessages().size());
+		assertEquals(1, messageStore2.getMessageGroup("group2").getMessages().size());
+
+		assertEquals(2, messageStore1.getMessageCount());
+		assertEquals(2, messageStore2.getMessageCount());
+
+	}
+
+	@Test
+	public void testRegionSetToMessageGroup() throws Exception {
+
+		messageStore1.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+
+		List<String> regions = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region1'", new RowMapper<String>(){
+
+			public String mapRow(ResultSet rs, int rowNum) throws SQLException {
+				return rs.getString("REGION");
+			}
+
+		});
+
+		assertEquals(1, regions.size());
+		assertEquals("region1", regions.get(0));
+
+		messageStore2.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+
+		List<String> regions2 = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region2'", new RowMapper<String>(){
+
+			public String mapRow(ResultSet rs, int rowNum) throws SQLException {
+				return rs.getString("REGION");
+			}
+
+		});
+
+		assertEquals(1, regions2.size());
+		assertEquals("region2", regions2.get(0));
+
+	}
+
+	@Test
+	public void testRemoveMessageGroup() throws Exception {
+
+		messageStore1.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+		messageStore1.addMessageToGroup("group2", MessageBuilder.withPayload("payload2").build());
+
+		messageStore2.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
+		messageStore2.addMessageToGroup("group2", MessageBuilder.withPayload("payload2").build());
+
+		messageStore1.removeMessageGroup("group1");
+
+		assertEquals(1, messageStore1.getMessageGroupCount());
+		assertEquals(2, messageStore2.getMessageGroupCount());
+
+	}
+}