Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2757 JdbcMessageStore fix removeMessageGroup

Browse Source 
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).
This commit is contained in:
Oleg Zhurakousky
2012-09-20 08:27:33 -04:00
committed by Gary Russell
parent 06e2f61969
commit b4af191f4f
2 changed files with 52 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<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},
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()){

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

@@ -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<String> 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());