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/JacksonChannelMessageStorePreparedStatementSetter.java

@@ -0,0 +1,133 @@
+/*
+ * 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.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.databind.ObjectMapper;
+
+import org.springframework.integration.support.json.JacksonMessagingUtils;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ChannelMessageStorePreparedStatementSetter} implementation that uses Jackson
+ * to serialize {@link Message} objects to JSON format 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.
+ * <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>
+ * The {@link ObjectMapper} is configured using {@link JacksonMessagingUtils#messagingAwareMapper(String...)}
+ * which includes custom serializers/deserializers for Spring Integration message types
+ * and embeds class type information for secure deserialization.
+ * <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 JacksonChannelMessageStorePreparedStatementSetter());
+ *     store.setMessageRowMapper(
+ *         new JacksonMessageRowMapper("com.example"));
+ *
+ *     return store;
+ * }
+ * }</pre>
+ *
+ * @author Yoobin Yoon
+ *
+ * @since 7.0
+ */
+public class JacksonChannelMessageStorePreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {
+
+	private final ObjectMapper objectMapper;
+
+	/**
+	 * Create a new {@link JacksonChannelMessageStorePreparedStatementSetter} with the
+	 * default trusted packages from {@link JacksonMessagingUtils#DEFAULT_TRUSTED_PACKAGES}.
+	 * <p>
+	 * This constructor is suitable when you only need to serialize standard Spring Integration
+	 * and Java classes. Custom payload types will require their package to be added to the
+	 * corresponding {@link JacksonMessageRowMapper}.
+	 */
+	public JacksonChannelMessageStorePreparedStatementSetter() {
+		super();
+		this.objectMapper = JacksonMessagingUtils.messagingAwareMapper();
+	}
+
+	/**
+	 * Create a new {@link JacksonChannelMessageStorePreparedStatementSetter} with a
+	 * custom {@link ObjectMapper}.
+	 * <p>
+	 * This constructor allows full control over the JSON serialization configuration.
+	 * The provided mapper should be configured appropriately for Message serialization,
+	 * typically using {@link JacksonMessagingUtils#messagingAwareMapper(String...)}.
+	 * <p>
+	 * <b>Note:</b> The same ObjectMapper configuration should be used in the corresponding
+	 * {@link JacksonMessageRowMapper} for consistent serialization and deserialization.
+	 * @param objectMapper the {@link ObjectMapper} to use for JSON serialization
+	 */
+	public JacksonChannelMessageStorePreparedStatementSetter(ObjectMapper objectMapper) {
+		super();
+		Assert.notNull(objectMapper, "'objectMapper' must not be null");
+		this.objectMapper = objectMapper;
+	}
+
+	@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.objectMapper.writeValueAsString(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 (JacksonException ex) {
+			throw new SQLException("Failed to serialize message to JSON: " + requestMessage, ex);
+		}
+	}
+
+}

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

@@ -0,0 +1,107 @@
+/*
+ * 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.sql.ResultSet;
+import java.sql.SQLException;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.core.type.TypeReference;
+import tools.jackson.databind.ObjectMapper;
+
+import org.springframework.integration.support.json.JacksonMessagingUtils;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.messaging.Message;
+
+/**
+ * A {@link RowMapper} implementation that deserializes {@link Message} objects from
+ * JSON format stored in the database.
+ * <p>
+ * This mapper works in conjunction with {@link JacksonChannelMessageStorePreparedStatementSetter}
+ * 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 Jackson to deserialize JSON strings from the MESSAGE_CONTENT column.
+ * <p>
+ * The {@link ObjectMapper} is configured using {@link JacksonMessagingUtils#messagingAwareMapper(String...)}
+ * which 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 JacksonChannelMessageStorePreparedStatementSetter());
+ *     store.setMessageRowMapper(
+ *         new JacksonMessageRowMapper("com.example"));
+ *
+ *     return store;
+ * }
+ * }</pre>
+ *
+ * @author Yoobin Yoon
+ *
+ * @since 7.0
+ */
+public class JacksonMessageRowMapper implements RowMapper<Message<?>> {
+
+	private final ObjectMapper objectMapper;
+
+	/**
+	 * Create a new {@link JacksonMessageRowMapper} with trusted packages for deserialization.
+	 * @param trustedPackages the packages to trust for deserialization
+	 */
+	public JacksonMessageRowMapper(String... trustedPackages) {
+		this.objectMapper = JacksonMessagingUtils.messagingAwareMapper(trustedPackages);
+	}
+
+	/**
+	 * Create a new {@link JacksonMessageRowMapper} with a custom {@link ObjectMapper}.
+	 * @param objectMapper the ObjectMapper configured for message serialization
+	 */
+	public JacksonMessageRowMapper(ObjectMapper objectMapper) {
+		this.objectMapper = objectMapper;
+	}
+
+	@Override
+	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.objectMapper.readValue(json, new TypeReference<Message<?>>() {
+
+			});
+		}
+		catch (JacksonException ex) {
+			throw new SQLException(
+					"Failed to deserialize message from JSON at row " + rowNum + ". "
+							+ "Ensure the JSON was created by JacksonChannelMessageStorePreparedStatementSetter "
+							+ "and contains proper @class type information.",
+					ex);
+		}
+	}
+
+}