GH-9312: Add JSON support for JdbcChannelMessageStore

Fixes: #9312

- Add `JsonChannelMessageStorePreparedStatementSetter` for serialization
- Add `JsonMessageRowMapper` for deserialization with trusted package validation
- Support PostgreSQL (JSONB), MySQL (JSON), and H2 (CLOB) databases
- Add comprehensive test coverage and documentation
- Document the feature

Signed-off-by: Yoobin Yoon <[email protected]>

Author: yybmion

build.gradle

@@ -686,16 +686,14 @@ project('spring-integration-jdbc') {
         testImplementation "org.apache.derby:derbytools:$derbyVersion"
         testImplementation "org.apache.derby:derbyclient:$derbyVersion"
         testImplementation "org.postgresql:postgresql:$postgresVersion"
-        testImplementation "com.mysql:mysql-connector-j:$mysqlVersion"
-        testImplementation("org.apache.commons:commons-dbcp2:$commonsDbcp2Version") {
-            exclude group: 'commons-logging'
-        }
+        testImplementation "org.apache.commons:commons-dbcp2:$commonsDbcp2Version"
         testImplementation 'org.testcontainers:testcontainers-mysql'
         testImplementation 'org.testcontainers:testcontainers-postgresql'
         testImplementation 'org.testcontainers:testcontainers-oracle-xe'
+		testImplementation 'tools.jackson.core:jackson-databind'
 
-        testRuntimeOnly 'tools.jackson.core:jackson-databind'
         testRuntimeOnly "com.oracle.database.jdbc:ojdbc11:$oracleVersion"
+		testRuntimeOnly "com.mysql:mysql-connector-j:$mysqlVersion"
     }
 }
 

spring-integration-core/src/main/java/org/springframework/integration/support/json/JacksonJsonUtils.java

@@ -83,7 +83,7 @@ private JacksonJsonUtils() {
 	 * @throws IllegalStateException if an implementation is not available.
 	 * @since 4.3.10
 	 */
-	public static ObjectMapper messagingAwareMapper(String... trustedPackages) {
+	public static ObjectMapper messagingAwareMapper(String @Nullable ... trustedPackages) {
 		if (JacksonPresent.isJackson2Present()) {
 			ObjectMapper mapper = new Jackson2JsonObjectMapper().getObjectMapper();
 

spring-integration-core/src/main/java/org/springframework/integration/support/json/JsonObjectMapperProvider.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.support.json;
 
+import org.jspecify.annotations.Nullable;
+
 /**
  * Simple factory to provide {@linkplain JsonObjectMapper}
  * instances based on jackson-databind lib in the classpath.
@@ -53,6 +55,25 @@ else if (JacksonPresent.isJackson2Present()) {
 		}
 	}
 
+	/**
+	 * Return an object mapper (if available) aware of messaging classes to (de)serialize.
+	 * @return the mapper.
+	 * @throws IllegalStateException if an implementation is not available.
+	 * @since 7.0
+	 */
+	@SuppressWarnings("removal")
+	public static JsonObjectMapper<?, ?> newMessagingAwareInstance(String @Nullable ... trustedPackages) {
+		if (JacksonPresent.isJackson3Present()) {
+			return new JacksonJsonObjectMapper(JacksonMessagingUtils.messagingAwareMapper(trustedPackages));
+		}
+		else if (JacksonPresent.isJackson2Present()) {
+			return new Jackson2JsonObjectMapper(JacksonJsonUtils.messagingAwareMapper(trustedPackages));
+		}
+		else {
+			throw new IllegalStateException("No jackson-databind.jar is present in the classpath.");
+		}
+	}
+
 	/**
 	 * Returns true if a supported JSON implementation is on the class path.
 	 * @return true if {@link #newInstance()} will return a mapper.

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

@@ -52,6 +52,7 @@
 import org.springframework.integration.support.converter.AllowListDeserializingConverter;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.jmx.export.annotation.ManagedAttribute;
@@ -89,6 +90,7 @@
  * @author Trung Pham
  * @author Johannes Edmeier
  * @author Ngoc Nhan
+ * @author Yoobin Yoon
  *
  * @since 2.2
  */
@@ -109,7 +111,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
 	 */
 	public static final String DEFAULT_TABLE_PREFIX = "INT_";
 
-	private enum Query {
+	protected enum Query {
 		CREATE_MESSAGE,
 		COUNT_GROUPS,
 		GROUP_SIZE,
@@ -148,7 +150,7 @@ private enum Query {
 	private SerializingConverter serializer;
 
 	@SuppressWarnings("NullAway.Init")
-	private MessageRowMapper messageRowMapper;
+	private RowMapper<Message<?>> messageRowMapper;
 
 	@SuppressWarnings("NullAway.Init")
 	private ChannelMessageStorePreparedStatementSetter preparedStatementSetter;
@@ -208,7 +210,7 @@ public void setDeserializer(Deserializer<? extends Message<?>> deserializer) {
 
 	/**
 	 * Add patterns for packages/classes that are allowed to be deserialized. A class can
-	 * be fully qualified or a wildcard '*' is allowed at the beginning or end of the
+	 * be fully qualified, or a wildcard '*' is allowed at the beginning or end of the
 	 * class name. Examples: {@code com.foo.*}, {@code *.MyClass}.
 	 * @param patterns the patterns.
 	 * @since 5.4
@@ -232,18 +234,18 @@ public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
 	}
 
 	/**
-	 * Allow for passing in a custom {@link MessageRowMapper}. The {@link MessageRowMapper}
-	 * is used to convert the selected database row representing the persisted
-	 * message into the actual {@link Message} object.
+	 * Allow for passing in a custom {@link RowMapper} for {@link Message}.
+	 * The {@link RowMapper} is used to convert the selected database row
+	 * representing the persisted message into the actual {@link Message} object.
 	 * @param messageRowMapper Must not be null
 	 */
-	public void setMessageRowMapper(MessageRowMapper messageRowMapper) {
-		Assert.notNull(messageRowMapper, "The provided MessageRowMapper must not be null.");
+	public void setMessageRowMapper(RowMapper<Message<?>> messageRowMapper) {
+		Assert.notNull(messageRowMapper, "The provided RowMapper must not be null.");
 		this.messageRowMapper = messageRowMapper;
 	}
 
 	/**
-	 * Set a {@link ChannelMessageStorePreparedStatementSetter} to insert message into the database.
+	 * Set a {@link ChannelMessageStorePreparedStatementSetter} to insert a message into the database.
 	 * @param preparedStatementSetter {@link ChannelMessageStorePreparedStatementSetter} to use.
 	 * Must not be null
 	 * @since 5.0
@@ -259,7 +261,7 @@ public void setPreparedStatementSetter(ChannelMessageStorePreparedStatementSette
 	 * The {@link JdbcChannelMessageStore} provides the SQL queries to retrieve messages from
 	 * the database. See the JavaDocs {@link ChannelMessageStoreQueryProvider} (all known
 	 * implementing classes) to see those implementations provided by the framework.
-	 * <p> You can provide your own query implementations, if you need to support additional
+	 * <p> You can provide your own query implementations if you need to support additional
 	 * databases and/or need to fine-tune the queries for your requirements.
 	 * @param channelMessageStoreQueryProvider Must not be null.
 	 */
@@ -313,7 +315,7 @@ public void setTablePrefix(String tablePrefix) {
 	 * using a task executor.</p>
 	 * <p>The issue is that the {@link #pollMessageFromGroup(Object)} looks for the
 	 * oldest entry for a giving channel (groupKey) and region ({@link #setRegion(String)}).
-	 * If you do that with multiple threads and you are using transactions, other
+	 * If you do that with multiple threads, and you are using transactions, other
 	 * threads may be waiting for that same locked row.</p>
 	 * <p>If using the provided
 	 * {@link org.springframework.integration.jdbc.store.channel.OracleChannelMessageStoreQueryProvider},
@@ -323,7 +325,7 @@ public void setTablePrefix(String tablePrefix) {
 	 * message id in an in-memory collection for the duration of processing. With
 	 * that, any polling threads will explicitly exclude those messages from
 	 * being polled.</p>
-	 * <p>For this to work, you must setup the corresponding
+	 * <p>For this to work, you must set up the corresponding
 	 * {@link org.springframework.integration.transaction.TransactionSynchronizationFactory}:</p>
 	 * <pre class="code">
 	 * {@code
@@ -388,7 +390,7 @@ protected MessageGroupFactory getMessageGroupFactory() {
 	 * Check mandatory properties ({@link DataSource} and
 	 * {@link #setChannelMessageStoreQueryProvider(ChannelMessageStoreQueryProvider)}). If no {@link MessageRowMapper}
 	 * and {@link ChannelMessageStorePreparedStatementSetter} was explicitly set using
-	 * {@link #setMessageRowMapper(MessageRowMapper)} and
+	 * {@link #setMessageRowMapper(RowMapper)} and
 	 * {@link #setPreparedStatementSetter(ChannelMessageStorePreparedStatementSetter)}  respectively, the default
 	 * {@link MessageRowMapper} and {@link ChannelMessageStorePreparedStatementSetter} will be instantiated using the
 	 * specified {@link #deserializer}.
@@ -493,7 +495,7 @@ public MessageGroup getMessageGroup(Object groupId) {
 	}
 
 	/**
-	 * Return the number of message groups in the store for configured region.
+	 * Return the number of message groups in the store for a configured region.
 	 * @return The message group count.
 	 */
 	@ManagedAttribute
@@ -564,8 +566,8 @@ private boolean isSingleStatementForPoll() {
 	 * This method executes a call to the DB to get the oldest Message in the
 	 * MessageGroup which in the context of the {@link JdbcChannelMessageStore}
 	 * means the channel identifier.
-	 * @param groupIdKey String representation of message group (Channel) ID
-	 * @return a message; could be null if query produced no Messages
+	 * @param groupIdKey String representation of a message group (Channel) ID
+	 * @return a message; could be null if a query produced no Messages
 	 */
 	protected @Nullable Message<?> doPollForMessage(String groupIdKey) {
 		NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(this.jdbcTemplate);

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

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2025-present 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.jdbc.store.channel;
+
+import java.io.IOException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+
+import org.springframework.integration.support.json.JsonObjectMapper;
+import org.springframework.integration.support.json.JsonObjectMapperProvider;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ChannelMessageStorePreparedStatementSetter} implementation that uses JSON
+ * serialization for {@link Message} objects instead of Java serialization.
+ * <p>
+ * By default, this implementation stores the entire message (including headers and payload) as JSON,
+ * with type information embedded using Jackson's {@code @class} property for proper deserialization.
+ * <p>
+ * <b>IMPORTANT:</b> JSON serialization exposes message content in text format in the database.
+ * Ensure proper database access controls and encryption for sensitive data.
+ * Consider the security implications before using this in production with sensitive information.
+ * <p>
+ * <b>Database Requirements:</b>
+ * This implementation requires modifying the MESSAGE_CONTENT column to a text-based type:
+ * <ul>
+ *     <li>PostgreSQL: Change from {@code BYTEA} to {@code JSONB}</li>
+ *     <li>MySQL: Change from {@code BLOB} to {@code JSON}</li>
+ *     <li>H2: Change from {@code LONGVARBINARY} to {@code CLOB}</li>
+ * </ul>
+ * See the reference documentation for schema migration instructions.
+ * <p>
+ * <b>Usage Example:</b>
+ * <pre>{@code
+ * &#64;Bean
+ * JdbcChannelMessageStore messageStore(DataSource dataSource) {
+ *     JdbcChannelMessageStore store = new JdbcChannelMessageStore(dataSource);
+ *     store.setChannelMessageStoreQueryProvider(new PostgresChannelMessageStoreQueryProvider());
+ *
+ *     // Enable JSON serialization (requires schema modification)
+ *     store.setPreparedStatementSetter(
+ *         new JsonChannelMessageStorePreparedStatementSetter());
+ *     store.setMessageRowMapper(
+ *         new JsonMessageRowMapper("com.example"));
+ *
+ *     return store;
+ * }
+ * }</pre>
+ *
+ * @author Yoobin Yoon
+ * @author Artem Bilan
+ *
+ * @since 7.0
+ */
+public class JsonChannelMessageStorePreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {
+
+	private final JsonObjectMapper<?, ?> jsonObjectMapper;
+
+	/**
+	 * Create a new {@link JsonChannelMessageStorePreparedStatementSetter} with the
+	 * default {@link JsonObjectMapper} configured for Spring Integration message serialization.
+	 * <p>
+	 * This constructor is suitable when serializing standard Spring Integration
+	 * and Java classes. Custom payload types will require their package to be added to the
+	 * corresponding {@link JsonMessageRowMapper}.
+	 */
+	public JsonChannelMessageStorePreparedStatementSetter() {
+		this(JsonObjectMapperProvider.newMessagingAwareInstance());
+	}
+
+	/**
+	 * Create a new {@link JsonChannelMessageStorePreparedStatementSetter} with a
+	 * custom {@link JsonObjectMapper}.
+	 * <p>
+	 * This constructor allows full control over the JSON serialization configuration.
+	 * <p>
+	 * <b>Note:</b> The same JsonObjectMapper configuration should be used in the corresponding
+	 * {@link JsonMessageRowMapper} for consistent serialization and deserialization.
+	 * @param jsonObjectMapper the {@link JsonObjectMapper} to use for JSON serialization
+	 */
+	public JsonChannelMessageStorePreparedStatementSetter(JsonObjectMapper<?, ?> jsonObjectMapper) {
+		super();
+		Assert.notNull(jsonObjectMapper, "'jsonObjectMapper' must not be null");
+		this.jsonObjectMapper = jsonObjectMapper;
+	}
+
+	@Override
+	public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
+			Object groupId, String region, boolean priorityEnabled) throws SQLException {
+
+		super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
+
+		try {
+			String json = this.jsonObjectMapper.toJson(requestMessage);
+
+			String dbProduct = preparedStatement.getConnection().getMetaData().getDatabaseProductName();
+
+			if ("PostgreSQL".equalsIgnoreCase(dbProduct)) {
+				preparedStatement.setObject(6, json, Types.OTHER);
+			}
+			else {
+				preparedStatement.setString(6, json);
+			}
+		}
+		catch (IOException ex) {
+			throw new SQLException("Failed to serialize message to JSON: " + requestMessage, ex);
+		}
+	}
+
+}