Messages deleted too early in JdbcMessageStore when using a message in multiple groups

[nschoellhorn]

In what version(s) of Spring Integration are you seeing this issue?

6.1.2

Describe the bug

When using a JdbcMessageStore and one group completes (or expires), the group is removed. This removal also removes the messages that were linked to that group, even when the same message is still linked to other groups.
The link (int_group_to_message) for that other group is not removed, so there's a mapping entry to a message that no longer exists in the database. Also, the size attribute of the other group is not updated, so it still shows a count that suggests the message is still present, while it can't be retrieved via getMessages() or streamMessages()

To Reproduce

Set up two flows listening on different channels. Both flows have an aggregator and use the JdbcMessageStore. Define different release strategies for each flow, but make sure there is an overlap in the "required messages" for each flow.
Now send the message that is required in both flows to both channels. After that, send a message that completes one of the groups. Then send a message that should complete the other group. It will not complete, since the first message is now missing from the group.

Expected behavior

The message should only be deleted from the int_message table, when it's not linked to any other group.

Sample
https://github.com/nschoellhorn/spring-integration-jdbc-repro

In SpringIntegrationJdbcReproApplication, I have implemented a CommandLineRunner that simulates the flow of events/messages as described above.
The flow setup can be found in Flows.kt

The Release strategy has some logging that should visualize the problem:

If my assumption is correct that this is a bug, I would be happy to assist in providing a patch to achieve the expected behavior.

[artembilan]

Thank you for the report!

I didn't run your application, but I can confirm that logically we really have a problem with many-to-one.
I believe that this query has a flaw:

				DELETE from INT_MESSAGE
				where MESSAGE_ID in (SELECT MESSAGE_ID from INT_GROUP_TO_MESSAGE where GROUP_KEY = ? and REGION = ?)
					and REGION = ?

The logic in the JdbcMessageStore is like this:

	@Override
	public void removeMessageGroup(Object groupId) {
		String groupKey = getKey(groupId);

		this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGES_FROM_GROUP), groupKey, this.region, this.region);

		if (logger.isDebugEnabled()) {
			logger.debug("Removing relationships for the group with group key=" + groupKey);
		}
		this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), groupKey, this.region);

		if (logger.isDebugEnabled()) {
			logger.debug("Deleting messages with group key=" + groupKey);
		}

		this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), groupKey, this.region);
	}

So, we delete messages from the INT_MESSAGE by their link to the current group.
Then we delete those links and only after that a record for a group by itself.

I'm not sure yet how to fix it, but this in indeed the problem when same message is sent to different aggregators on the same message store.

As a workaround I suggest to have different JdbcMessageStore instances with their own region property.
This way messages sent to these different stores will be partitioned to their own region.
See the PK on that table:

CREATE TABLE INT_MESSAGE  (
	MESSAGE_ID CHAR(36) NOT NULL,
	REGION VARCHAR(100) NOT NULL,
	CREATED_DATE TIMESTAMP NOT NULL,
	MESSAGE_BYTES BYTEA,
	constraint INT_MESSAGE_PK primary key (MESSAGE_ID, REGION)
);

Any your thoughts what would we do as a fix for this problem?

Thanks

[artembilan]

So, I have reproduced it with this simple test:

	@Test
	public void testSameMessageTwoGroups() {
		GenericMessage<String> testMessage = new GenericMessage<>("test data");

		messageStore.addMessageToGroup("1", testMessage);
		messageStore.addMessageToGroup("2", testMessage);

		messageStore.removeMessageGroup("1");

		assertThat(messageStore.getMessageCount()).isEqualTo(1);
	}

Where that message indeed has to remain in the INT_MESSAGE because of a second group.

I found solution like this:

				DELETE from %PREFIX%MESSAGE
				where MESSAGE_ID in (
								SELECT MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE where GROUP_KEY = ? and REGION = ?
												and MESSAGE_ID not in (
																SELECT MESSAGE_ID from %PREFIX%GROUP_TO_MESSAGE
																			where GROUP_KEY != ? and REGION = ?)
								)
					and REGION = ?

But my SQL skills are lacking experience 😄

[artembilan]

I've decided to push my fix for your and others review: #8733.

I'm fully open for other suggestions.

Thanks

[nschoellhorn]

Hi,

first of all, thanks for the quick reply. This really helps, since the bug was a show stopper for us currently. Hence, I was also working on a patch in parallel. I came to the same solution as you did, but I also think it's not the full solution. It only works when expireGroupsUponCompletion is set to true.

When it's set to false, the code executed in AggregatingMessageHandler is different. It uses removeMessagesFromGroup() here, which would still delete the messages.

This gets a bit more interesting now. We could adapt the query that's being used there in the same way (where not in ...), but the query is also used in removeMessage(). The contract (or rather the documentation?!) of removeMessage() states that it should remove the message from the store, not from a group. So at this point, it's not clear to me if there's a conceptual issue with multiple JdbcMessageStores in one application in general?

To make it clear, my initial issue (message deleted when expireGroupsUponCompletion is set to true) is resolved with the fix you provided but I think it's worth discussing the full picture here since it seems related to me.