Add Jackson JSON serialization for JdbcChannelMessageStore

- Add JacksonChannelMessageStorePreparedStatementSetter for serialization
- Add JacksonMessageRowMapper for deserialization with trusted package validation
- Support PostgreSQL (JSONB), MySQL (JSON), and H2 (CLOB) databases
- Add comprehensive test coverage and documentation

Fixes: gh-9312

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

Author: yybmion

build.gradle

@@ -679,6 +679,7 @@ project('spring-integration-jdbc') {
     dependencies {
         api 'org.springframework:spring-jdbc'
         optionalApi "org.postgresql:postgresql:$postgresVersion"
+        optionalApi 'tools.jackson.core:jackson-databind'
 
         testImplementation "com.h2database:h2:$h2Version"
         testImplementation "org.hsqldb:hsqldb:$hsqldbVersion"

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
  */
@@ -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;
@@ -232,13 +234,13 @@ 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;
 	}
 
@@ -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}.

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.JacksonJsonObjectMapper;
+import org.springframework.integration.support.json.JacksonMessagingUtils;
+import org.springframework.integration.support.json.JsonObjectMapper;
+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>
+ * 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
+ *
+ * @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(new JacksonJsonObjectMapper(JacksonMessagingUtils.messagingAwareMapper()));
+	}
+
+	/**
+	 * 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); // NOSONAR magic number
+			}
+			else {
+				preparedStatement.setString(6, json); // NOSONAR magic number
+			}
+		}
+		catch (IOException ex) {
+			throw new SQLException("Failed to serialize message to JSON: " + requestMessage, ex);
+		}
+	}
+
+}

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

@@ -0,0 +1,129 @@
+/*
+ * 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.ResultSet;
+import java.sql.SQLException;
+
+import org.springframework.integration.support.json.JacksonJsonObjectMapper;
+import org.springframework.integration.support.json.JacksonMessagingUtils;
+import org.springframework.integration.support.json.JsonObjectMapper;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link RowMapper} implementation that deserializes {@link Message} objects from
+ * JSON format stored in the database.
+ * <p>
+ * This mapper works in conjunction with {@link JsonChannelMessageStorePreparedStatementSetter}
+ * to provide JSON serialization for Spring Integration's JDBC Channel Message Store.
+ * <p>
+ * Unlike the default {@link MessageRowMapper} which uses Java serialization,
+ * this implementation uses JSON to deserialize message strings from the MESSAGE_CONTENT column.
+ * <p>
+ * The underlying {@link JsonObjectMapper} validates all deserialized classes against a
+ * trusted package list to prevent security vulnerabilities.
+ * <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
+ *     store.setPreparedStatementSetter(
+ *         new JsonChannelMessageStorePreparedStatementSetter());
+ *     store.setMessageRowMapper(
+ *         new JsonMessageRowMapper("com.example"));
+ *
+ *     return store;
+ * }
+ * }</pre>
+ *
+ * @author Yoobin Yoon
+ *
+ * @since 7.0
+ */
+public class JsonMessageRowMapper implements RowMapper<Message<?>> {
+
+	private final JsonObjectMapper<?, ?> jsonObjectMapper;
+
+	/**
+	 * Create a new {@link JsonMessageRowMapper} with the default {@link JsonObjectMapper}
+	 * configured for Spring Integration message deserialization.
+	 * <p>
+	 * This constructor is suitable when deserializing standard Spring Integration
+	 * and Java classes. Custom payload types will require their package to be
+	 * specified using {@link #JsonMessageRowMapper(String...)}.
+	 */
+	public JsonMessageRowMapper() {
+		this(new JacksonJsonObjectMapper(JacksonMessagingUtils.messagingAwareMapper()));
+	}
+
+	/**
+	 * Create a new {@link JsonMessageRowMapper} with additional trusted packages
+	 * for deserialization.
+	 * <p>
+	 * The provided packages are appended to the default trusted packages,
+	 * enabling deserialization of custom payload types while maintaining security.
+	 * @param trustedPackages the additional packages to trust for deserialization
+	 */
+	public JsonMessageRowMapper(String... trustedPackages) {
+		this(new JacksonJsonObjectMapper(
+				JacksonMessagingUtils.messagingAwareMapper(trustedPackages)));
+	}
+
+	/**
+	 * Create a new {@link JsonMessageRowMapper} with a custom {@link JsonObjectMapper}.
+	 * <p>
+	 * This constructor allows full control over the JSON deserialization configuration.
+	 * <p>
+	 * <b>Note:</b> The same JsonObjectMapper configuration should be used in the corresponding
+	 * {@link JsonChannelMessageStorePreparedStatementSetter} for consistent
+	 * serialization and deserialization.
+	 * @param jsonObjectMapper the {@link JsonObjectMapper} to use for JSON deserialization
+	 */
+	public JsonMessageRowMapper(JsonObjectMapper<?, ?> jsonObjectMapper) {
+		Assert.notNull(jsonObjectMapper, "'jsonObjectMapper' must not be null");
+		this.jsonObjectMapper = jsonObjectMapper;
+	}
+
+	@Override
+	@SuppressWarnings("NullAway")
+	public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
+		try {
+			String json = rs.getString("MESSAGE_CONTENT");
+
+			if (json == null) {
+				throw new SQLException("MESSAGE_CONTENT column is null at row " + rowNum);
+			}
+
+			return this.jsonObjectMapper.fromJson(json, Message.class);
+		}
+		catch (IOException ex) {
+			throw new SQLException(
+					"Failed to deserialize message from JSON at row " + rowNum + ". "
+							+ "Ensure the JSON was created by JsonChannelMessageStorePreparedStatementSetter "
+							+ "and contains proper @class type information.",
+					ex);
+		}
+	}
+
+}