JdbcMessageStore doesn't close database connections after streaming stored messages

[Seregy]

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

5.5.10

Describe the bug

Usage of JdbcMessageStore eventually leads to the database connection pool depletion as the connections are not getting properly closed after streaming the messages. After a while, the application stops functioning correctly, as the database connection pool never receives those connections back and treats them as active.

This seems to happen on:

• message group expiration via the configured timeout
• explicit execution of a MessageGroupStoreReaper instance
• streaming through the messages of a message group via getMessages().stream()

The issue appears to be with the stream from PersistentMessageGroup#streamMessages not getting closed after the usage in certain cases.

To Reproduce

Set up the application with a limited database connection pool, configure an aggregation flow with an instance of JdbcMessageStore and a scheduled MessageGroupStoreReaper execution for that message store.

Expected behavior

Internal usages of the message store should properly release database connections when JdbcMessageStore is utilized.

If the current behavior of streaming on MessageGroup#getMessages is expected, the need for closing the stream should ideally be mentioned in the documentation.

Sample

Sample application

[artembilan]

Yeah... Looks like an omission on our side.
The JdbcTemplate states that clearly:

	/**
	 * Query given SQL to create a prepared statement from SQL and a list of
	 * arguments to bind to the query, mapping each row to a result object
	 * via a RowMapper, and turning it into an iterable and closeable Stream.
	 * @param sql the SQL query to execute
	 * @param rowMapper a callback that will map one object per row
	 * @param args arguments to bind to the query
	 * (leaving it to the PreparedStatement to guess the corresponding SQL type);
	 * may also contain {@link SqlParameterValue} objects which indicate not
	 * only the argument value but also the SQL type and optionally the scale
	 * @return the result Stream, containing mapped objects, needing to be
	 * closed once fully processed (e.g. through a try-with-resources clause)
	 * @throws DataAccessException if the query fails
	 * @since 5.3
	 */
	<T> Stream<T> queryForStream(String sql, RowMapper<T> rowMapper, @Nullable Object... args)
			throws DataAccessException;

Has to be closed via try-with-resources.

Will fix shortly.

Thank you!

[artembilan]

The problem is here PersistentMessageGroup.PersistentCollection.stream() which is used in the CollectionArgumentResolver as:

value = messages.stream()
						.map(Message::getPayload)
						.collect(Collectors.toList());

I wonder if there are more places like this...

[artembilan]

Well, we even mention that in docs: https://docs.spring.io/spring-integration/docs/current/reference/html/system-management.html#lazy-load-message-group

A returned Stream<Message<?>> must be closed in the end of processing, e.g. via auto-close by the try-with-resources.

So, this is just a bug in the framework code by itself 😄