INT-3846: SimpleMessageStore Improvements

JIRAs:
https://jira.spring.io/browse/INT-3830
https://jira.spring.io/browse/INT-3523
https://jira.spring.io/browse/INT-3846

* Fix the OOM condition, when we `release()` `UpperBound` independently of the previous `remove` result (https://jira.spring.io/browse/INT-3846)
* Fix "confuse" around `groupCapacity`, when we really didn't care about individual groups (https://jira.spring.io/browse/INT-3523)
* Add `upperBoundTimeout` to have a hook to wait some time for the empty slot in the store (https://jira.spring.io/browse/INT-3830)
* Fix some JavaDocs warnings
* Fix some typos

* Fix inconsistency in the `DelayHandler` around `removeMessageFromGroup` when `MS` is `SimpleMessageStore`
 * Remove `UpperBound.release()` operation from `SimpleMessageStore.removeGroup()`.
 The waiting process should worry about the new `UpperBound` instance.

Some other polishing

`tryAcquire` outside of the `lock`

Move `tryAcquire` within the `addMessageToGroup` outside of `lock`.
But do that only for groups which already exist.
For the new groups we have a fresh `UpperBound`, so no need to worry about dead lock and
 we can obtain  a permit immediately.

SimpleMessageGroup: BlockingQueue -> LinkedHashSet

`SimpleMessageStore`: use "unsynchonized" `SimpleMessageGroup`

Make some synchronization fixes according to the migration to the `LinkedHashSet`

Avoid extra `Collection`

`ResequencingMessageHandler`: compare `size()` of collections instead of `containsAll()`

Fix `ConcurrentModificationException` in the `AbstractKeyValueMessageStore`

Add `SimpleMessageStore.clearMessageGroup()`

Accept polishing and fix `RedisChannelMessageStoreTests`

`@Deprecated` `MessageGroupStore.removeMessageFromGroup()`

Fix some typos

Introduce `SimpleMessageGroupFactory`

Extract `MessageGroupFactory` and address PR comments

Polishing after rebase

JavaDocs and Reference Manual

Fix JavaDocs

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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
@@ -44,7 +44,6 @@ import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
 import org.springframework.integration.store.AbstractMessageGroupStore;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageStore;
-import org.springframework.integration.store.SimpleMessageGroup;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -79,6 +78,7 @@ import org.springframework.util.StringUtils;
  * @author Gunnar Hillert
  * @author Will Schipp
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 2.0
  */
@@ -169,8 +169,6 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 		}
 	}
 
-	public static final int DEFAULT_LONG_STRING_LENGTH = 2500;
-
 	/**
 	 * The name of the message header that stores a flag to indicate that the message has been saved. This is an
 	 * optimization for the put method.
@@ -427,10 +425,12 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 		final AtomicReference<Boolean> completeFlag = new AtomicReference<Boolean>();
 		final AtomicReference<Integer> lastReleasedSequenceRef = new AtomicReference<Integer>();
 
-		List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY), new Object[] { key, region }, mapper);
+		List<Message<?>> messages = jdbcTemplate.query(getQuery(Query.LIST_MESSAGES_BY_GROUP_KEY),
+				new Object[] { key, region }, mapper);
 
 		jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key, region},
 				new RowCallbackHandler() {
+
 					@Override
 					public void processRow(ResultSet rs) throws SQLException {
 						updateDate.set(rs.getTimestamp("UPDATED_DATE"));
@@ -441,6 +441,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 
 						lastReleasedSequenceRef.set(rs.getInt("LAST_RELEASED_SEQUENCE"));
 					}
+
 				});
 
 		if (createDate.get() == null && updateDate.get() == null) {
@@ -449,22 +450,23 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 					logger.warn("Missing group row for message id: " + message.getHeaders().getId());
 				}
 			}
-			return new SimpleMessageGroup(groupId);
+			return getMessageGroupFactory().create(groupId);
 		}
 
 		long timestamp = createDate.get().getTime();
 		boolean complete = completeFlag.get();
-
-		SimpleMessageGroup messageGroup = new SimpleMessageGroup(messages, groupId, timestamp, complete);
-		messageGroup.setLastModified(updateDate.get().getTime());
-
+		long lastModified = updateDate.get().getTime();
 		int lastReleasedSequenceNumber = lastReleasedSequenceRef.get();
-		messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber);
 
+		MessageGroup messageGroup = getMessageGroupFactory()
+				.create(messages, groupId, timestamp, complete);
+		messageGroup.setLastModified(lastModified);
+		messageGroup.setLastReleasedMessageSequenceNumber(lastReleasedSequenceNumber);
 		return messageGroup;
 	}
 
 	@Override
+	@Deprecated
 	public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
 		final String groupKey = getKey(groupId);
 		final String messageId = getKey(messageToRemove.getHeaders().getId());
@@ -499,22 +501,22 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
 				messages,
 				getRemoveBatchSize(),
 				new ParameterizedPreparedStatementSetter<Message<?>>() {
-			@Override
-			public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
-				ps.setString(1, groupKey);
-				ps.setString(2, getKey(messageToRemove.getHeaders().getId()));
-				ps.setString(3, region);
-			}
+					@Override
+					public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
+						ps.setString(1, groupKey);
+						ps.setString(2, getKey(messageToRemove.getHeaders().getId()));
+						ps.setString(3, region);
+					}
 		});
 		jdbcTemplate.batchUpdate(getQuery(Query.DELETE_MESSAGE),
 				messages,
 				getRemoveBatchSize(),
 				new ParameterizedPreparedStatementSetter<Message<?>>() {
-			@Override
-			public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
-				ps.setString(1, getKey(messageToRemove.getHeaders().getId()));
-				ps.setString(2, region);
-			}
+					@Override
+					public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
+						ps.setString(1, getKey(messageToRemove.getHeaders().getId()));
+						ps.setString(2, region);
+					}
 		});
 		this.updateMessageGroup(groupKey);
 	}

spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 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
@@ -25,7 +25,6 @@ import java.util.UUID;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
-
 import javax.sql.DataSource;
 
 import org.apache.commons.logging.Log;
@@ -49,10 +48,11 @@ import org.springframework.integration.jdbc.store.channel.MySqlChannelMessageSto
 import org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider;
 import org.springframework.integration.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupFactory;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.PriorityCapableChannelMessageStore;
-import org.springframework.integration.store.SimpleMessageGroup;
+import org.springframework.integration.store.SimpleMessageGroupFactory;
 import org.springframework.integration.support.DefaultMessageBuilderFactory;
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.utils.IntegrationUtils;
@@ -151,6 +151,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 
 	private volatile Map<String, String> queryCache = new HashMap<String, String>();
 
+	private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory();
+
 	private boolean usingIdCache = false;
 
 	private boolean priorityEnabled;
@@ -360,6 +362,22 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 		return this.priorityEnabled;
 	}
 
+	/**
+	 * Specify the {@link MessageGroupFactory} to create {@link MessageGroup} object where
+	 * it is necessary.
+	 * Defaults to {@link SimpleMessageGroupFactory}.
+	 * @param messageGroupFactory the {@link MessageGroupFactory} to use.
+	 * @since 4.3
+	 */
+	public void setMessageGroupFactory(MessageGroupFactory messageGroupFactory) {
+		Assert.notNull(messageGroupFactory, "'messageGroupFactory' must not be null");
+		this.messageGroupFactory = messageGroupFactory;
+	}
+
+	protected MessageGroupFactory getMessageGroupFactory() {
+		return this.messageGroupFactory;
+	}
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -468,7 +486,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 	 */
 	@Override
 	public MessageGroup getMessageGroup(Object groupId) {
-		return new SimpleMessageGroup(groupId);
+		return getMessageGroupFactory().create(groupId);
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -243,7 +243,7 @@ public class JdbcMessageStoreTests {
 		String groupId = "X";
 		Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
 		messageStore.addMessageToGroup(groupId, message);
-		messageStore.removeMessageFromGroup(groupId, message);
+		messageStore.removeMessagesFromGroup(groupId, message);
 		MessageGroup group = messageStore.getMessageGroup(groupId);
 		assertEquals(0, group.size());
 	}
@@ -532,7 +532,7 @@ public class JdbcMessageStoreTests {
 		messageStore.completeGroup(messageGroup.getGroupId());
 		//now clear the messages
 		for (Message<?> message : messageGroup.getMessages()) {
-			messageStore.removeMessageFromGroup(groupId, message);
+			messageStore.removeMessagesFromGroup(groupId, message);
 		}//end for
 		//'add' the other message --> emulated by getting the messageGroup
 		messageGroup = messageStore.getMessageGroup(groupId);
@@ -540,6 +540,4 @@ public class JdbcMessageStoreTests {
 		assertTrue(messageGroup.isComplete());
 	}
 
-
-
 }

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/mysql/MySqlJdbcMessageStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -42,6 +42,7 @@ import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
@@ -84,10 +85,11 @@ import org.springframework.transaction.support.TransactionTemplate;
  * schema-mysql-5_6_4.sql
  *
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
-@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD)
+@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
 @Ignore
 public class MySqlJdbcMessageStoreTests {
 
@@ -111,16 +113,17 @@ public class MySqlJdbcMessageStoreTests {
 	public void afterTest() {
 		final JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
 		new TransactionTemplate(this.transactionManager).execute(new TransactionCallback<Void>() {
-		public Void doInTransaction(TransactionStatus status) {
-			final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
-			final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE");
-			final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
 
-			LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " +
-					"Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s",
-					deletedMessages, deletedGroupToMessageRows, deletedMessageGroups));
-			return null;
-		}
+			public Void doInTransaction(TransactionStatus status) {
+				final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
+				final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE");
+				final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
+
+				LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " +
+								"Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s",
+						deletedMessages, deletedGroupToMessageRows, deletedMessageGroups));
+				return null;
+			}
 		});
 	}
 
@@ -146,7 +149,7 @@ public class MySqlJdbcMessageStoreTests {
 
 	@Test
 	@Transactional
-	public void testWithMessageHistory() throws Exception{
+	public void testWithMessageHistory() throws Exception {
 
 		Message<?> message = new GenericMessage<String>("Hello");
 		DirectChannel fooChannel = new DirectChannel();
@@ -179,12 +182,14 @@ public class MySqlJdbcMessageStoreTests {
 	public void testSerializer() throws Exception {
 		// N.B. these serializers are not realistic (just for test purposes)
 		messageStore.setSerializer(new Serializer<Message<?>>() {
+
 			public void serialize(Message<?> object, OutputStream outputStream) throws IOException {
 				outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
 				outputStream.flush();
 			}
 		});
 		messageStore.setDeserializer(new Deserializer<GenericMessage<String>>() {
+
 			public GenericMessage<String> deserialize(InputStream inputStream) throws IOException {
 				BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
 				return new GenericMessage<String>(reader.readLine());
@@ -277,7 +282,7 @@ public class MySqlJdbcMessageStoreTests {
 		String groupId = "X";
 		Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
 		messageStore.addMessageToGroup(groupId, message);
-		messageStore.removeMessageFromGroup(groupId, message);
+		messageStore.removeMessagesFromGroup(groupId, message);
 		MessageGroup group = messageStore.getMessageGroup(groupId);
 		assertEquals(0, group.size());
 	}
@@ -297,7 +302,7 @@ public class MySqlJdbcMessageStoreTests {
 
 		String uuidGroupId = UUIDConverter.getUUID(groupId).toString();
 		assertTrue(template.queryForList(
-				"SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '"  + uuidGroupId + "'").size() == 0);
+				"SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0);
 	}
 
 	@Test
@@ -362,6 +367,7 @@ public class MySqlJdbcMessageStoreTests {
 		Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
 		messageStore.addMessageToGroup(groupId, message);
 		messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
+
 			public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
 				messageGroupStore.removeMessageGroup(group.getGroupId());
 			}
@@ -385,6 +391,7 @@ public class MySqlJdbcMessageStoreTests {
 		messageStore.setTimeoutOnIdle(true);
 		messageStore.addMessageToGroup(groupId, message);
 		messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
+
 			public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
 				messageGroupStore.removeMessageGroup(group.getGroupId());
 			}
@@ -511,10 +518,11 @@ public class MySqlJdbcMessageStoreTests {
 		assertNotNull(messageFromRegion2);
 
 		LOG.info("messageFromRegion1: " + messageFromRegion1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion1).getSequenceNumber());
-		LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " +new IntegrationMessageHeaderAccessor(messageFromRegion2).getSequenceNumber());
+		LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion2).getSequenceNumber());
 
 		assertEquals(Integer.valueOf(1), (Integer) messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
 		assertEquals(Integer.valueOf(2), (Integer) messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
 
 	}
+
 }