From b4af191f4f68b131e81a8d21ebcc903d9da6700b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 20 Sep 2012 08:27:33 -0400 Subject: [PATCH] INT-2757 JdbcMessageStore fix removeMessageGroup Fix removeMessageGroup logic to ensure it also executes delete on the join table so there are no dangling join rows. Add defensive logic to the getMessageGroup method to ensure it no longer throws an exception IF the scenario fixed with the previous comment still re-appears (and it can if MGS is not executed under TX). --- .../integration/jdbc/JdbcMessageStore.java | 26 ++++++++++-- .../jdbc/JdbcMessageStoreTests.java | 41 ++++++++++++++----- 2 files changed, 52 insertions(+), 15 deletions(-) diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index 020ebe4d98..da7b60b001 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -89,6 +89,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa REMOVE_MESSAGE_FROM_GROUP("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=? and MESSAGE_ID=?"), + REMOVE_GROUP_TO_MESSAGE_JOIN("DELETE from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"), + COUNT_ALL_MESSAGES_IN_GROUPS("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE"), COUNT_ALL_MESSAGES_IN_GROUP("SELECT COUNT(MESSAGE_ID) from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY=?"), @@ -403,7 +405,11 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa List> 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}, + if (messages.size() == 0){ + return new SimpleMessageGroup(groupId); + } + + jdbcTemplate.query(getQuery(Query.GET_GROUP_INFO), new Object[] { key}, new RowCallbackHandler() { public void processRow(ResultSet rs) throws SQLException { updateDate.set(rs.getTimestamp("UPDATED_DATE")); @@ -416,11 +422,14 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa } }); - if (messages.size() == 0){ + if (createDate.get() == null && updateDate.get() == null) { + if (logger.isWarnEnabled()) { + for (Message message : messages) { + logger.warn("Missing group row for message id: " + message.getHeaders().getId()); + } + } return new SimpleMessageGroup(groupId); } - Assert.state(createDate.get() != null, "Could not locate created date for groupId=" + groupId); - Assert.state(updateDate.get() != null, "Could not locate updated date for groupId=" + groupId); long timestamp = createDate.get().getTime(); boolean complete = completeFlag.get().booleanValue(); @@ -460,6 +469,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa this.removeMessage(messageIds); } + jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + if (logger.isDebugEnabled()){ + logger.debug("Removing relationships for the group with group key=" + groupKey); + } + ps.setString(1, groupKey); + } + }); + jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { if (logger.isDebugEnabled()){ diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index eb13322028..cfc9b9143f 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -16,6 +16,15 @@ package org.springframework.integration.jdbc; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; + import java.io.BufferedReader; import java.io.IOException; import java.io.InputStream; @@ -41,20 +50,12 @@ import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupCallback; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.util.UUIDConverter; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.transaction.annotation.Transactional; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNotSame; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertSame; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; - -import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; - /** * @author Dave Syer * @author Mark Fisher @@ -234,6 +235,24 @@ public class JdbcMessageStoreTests { assertEquals(0, group.size()); } + @Test + @Transactional + public void testRemoveMessageGroup() throws Exception { + JdbcTemplate template = new JdbcTemplate(dataSource); + template.afterPropertiesSet(); + String groupId = "X"; + + Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); + messageStore.addMessageToGroup(groupId, message); + messageStore.removeMessageGroup(groupId); + MessageGroup group = messageStore.getMessageGroup(groupId); + assertEquals(0, group.size()); + + String uuidGroupId = UUIDConverter.getUUID(groupId).toString(); + assertTrue(template.queryForList( + "SELECT * from INT_GROUP_TO_MESSAGE where GROUP_KEY = '" + uuidGroupId + "'").size() == 0); + } + @Test @Transactional public void testCompleteMessageGroup() throws Exception { @@ -279,7 +298,7 @@ public class JdbcMessageStoreTests { @Transactional public void testOrderInMessageGroup() throws Exception { String groupId = "X"; - + messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("foo").setCorrelationId(groupId).build()); Thread.sleep(1); messageStore.addMessageToGroup(groupId, MessageBuilder.withPayload("bar").setCorrelationId(groupId).build());