This version is still in development and is not considered stable yet. For the latest stable version, please use Spring Integration 6.5.3!

JDBC Channel Message Store JSON Serialization

Version 7.0 introduced JSON serialization support for JdbcChannelMessageStore. By default, Spring Integration uses Java serialization to store messages in the database. The new JSON serialization option provides an alternative serialization mechanism.

Security Consideration: JSON serialization stores message content as text in the database, which may expose sensitive data. Ensure proper database access controls, encryption at rest, and consider your organization’s data protection requirements before using JSON serialization in production environments.

Configuration

Two components are available for JSON (de)serialization:

JsonChannelMessageStorePreparedStatementSetter - Serializes messages to JSON
JsonMessageRowMapper - Deserializes messages from JSON

@Bean
public 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;
}

The string parameter ("com.example") specifies additional trusted packages for deserialization. These packages are appended to the default trusted packages (see Trusted Packages section). Only classes from trusted packages can be deserialized for security.

Database Schema Modification

JSON serialization requires modifying the database schema. The default schema with BLOB/BYTEA column types cannot be used for JSON serialization.

The MESSAGE_CONTENT column must be changed to a text-based type that can store JSON.

PostgreSQL
MySQL
H2
Other Databases

For PostgreSQL, the JSONB type can be used.

-- JSONB (enables JSON queries)
ALTER TABLE INT_CHANNEL_MESSAGE
ALTER COLUMN MESSAGE_CONTENT TYPE JSONB
USING MESSAGE_CONTENT::text::jsonb;

For MySQL, the JSON type can be used.

-- JSON type (enables JSON functions)
ALTER TABLE INT_CHANNEL_MESSAGE
MODIFY COLUMN MESSAGE_CONTENT JSON;

For H2 database, the CLOB type can be used.

ALTER TABLE INT_CHANNEL_MESSAGE
ALTER COLUMN MESSAGE_CONTENT CLOB;

For any database that supports large text columns (CLOB, TEXT, etc.), the MESSAGE_CONTENT column can be modified to an appropriate text type.

Example Schema for JSON Serialization

The following examples demonstrate how to create a dedicated table for JSON-based message storage.

PostgreSQL
MySQL
H2

CREATE TABLE JSON_CHANNEL_MESSAGE (
   MESSAGE_ID CHAR(36) NOT NULL,
   GROUP_KEY CHAR(36) NOT NULL,
   CREATED_DATE BIGINT NOT NULL,
   MESSAGE_PRIORITY BIGINT,
   MESSAGE_SEQUENCE BIGINT NOT NULL DEFAULT nextval('JSON_MESSAGE_SEQ'),
   MESSAGE_CONTENT JSONB, -- JSON message content
   REGION VARCHAR(100) NOT NULL,
   CONSTRAINT JSON_CHANNEL_MESSAGE_PK
      PRIMARY KEY (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);

CREATE TABLE JSON_CHANNEL_MESSAGE (
   MESSAGE_ID CHAR(36) NOT NULL,
   GROUP_KEY CHAR(36) NOT NULL,
   CREATED_DATE BIGINT NOT NULL,
   MESSAGE_PRIORITY BIGINT,
   MESSAGE_SEQUENCE BIGINT NOT NULL AUTO_INCREMENT UNIQUE,
   MESSAGE_CONTENT JSON, -- JSON message content
   REGION VARCHAR(100) NOT NULL,
   CONSTRAINT JSON_CHANNEL_MESSAGE_PK
      PRIMARY KEY (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);

CREATE TABLE JSON_CHANNEL_MESSAGE (
   MESSAGE_ID CHAR(36) NOT NULL,
   GROUP_KEY CHAR(36) NOT NULL,
   CREATED_DATE BIGINT NOT NULL,
   MESSAGE_PRIORITY BIGINT,
   MESSAGE_SEQUENCE BIGINT NOT NULL DEFAULT NEXT VALUE FOR JSON_MESSAGE_SEQ,
   MESSAGE_CONTENT CLOB, -- JSON message content
   REGION VARCHAR(100) NOT NULL,
   CONSTRAINT JSON_CHANNEL_MESSAGE_PK
      PRIMARY KEY (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);

JSON Structure

When using Jackson-based serialization, messages are stored with the following JSON structure using Jackson’s polymorphic type handling:

{
  "@class": "org.springframework.messaging.support.GenericMessage",
  "payload": {
    "@class": "com.example.OrderMessage",
    "orderId": "ORDER-12345",
    "amount": 1299.99
  },
  "headers": {
    "@class": "java.util.HashMap",
    "priority": ["java.lang.String", "HIGH"],
    "id": ["java.util.UUID", "a1b2c3d4-..."],
    "timestamp": ["java.lang.Long", 1234567890]
  }
}

The @class properties provide type information necessary for proper deserialization of polymorphic types.

Querying JSON Content (Optional)

If native JSON column types (PostgreSQL JSONB or MySQL JSON) are used, message content can be queried directly.

PostgreSQL JSONB Queries

-- Find messages by payload field
SELECT * FROM JSON_CHANNEL_MESSAGE
WHERE MESSAGE_CONTENT @> '{"payload": {"orderId": "ORDER-12345"}}';

-- Find high-priority messages
SELECT * FROM JSON_CHANNEL_MESSAGE
WHERE MESSAGE_CONTENT -> 'headers' @> '{"priority": ["java.lang.String", "HIGH"]}';

MySQL JSON Functions

-- Find messages by payload field
SELECT * FROM JSON_CHANNEL_MESSAGE
WHERE JSON_EXTRACT(MESSAGE_CONTENT, '$.payload.orderId') = 'ORDER-12345';

-- Find high-priority messages
SELECT * FROM JSON_CHANNEL_MESSAGE
WHERE JSON_EXTRACT(MESSAGE_CONTENT, '$.headers.priority[1]') = 'HIGH';

If TEXT or CLOB column types are used, these JSON-specific queries are not available. However, JSON serialization still works for storage and retrieval through Spring Integration.

Trusted Packages

The JacksonMessagingUtils.messagingAwareMapper() validates all deserialized classes against a trusted package list to prevent security vulnerabilities.

Default trusted packages include: - java.util - java.lang - org.springframework.messaging.support - org.springframework.integration.support - org.springframework.integration.message - org.springframework.integration.store - org.springframework.integration.history - org.springframework.integration.handler

Additional packages can be specified in the constructor and are appended to this list:

// Trust additional packages for custom payload types
new JsonMessageRowMapper("com.example.orders", "com.example.payments")

Custom JsonObjectMapper

For advanced scenarios, a custom JsonObjectMapper can be provided:

import org.springframework.integration.support.json.JacksonJsonObjectMapper;
import org.springframework.integration.support.json.JacksonMessagingUtils;
import tools.jackson.databind.ObjectMapper;
import tools.jackson.databind.SerializationFeature;

@Bean
public JdbcChannelMessageStore messageStore(DataSource dataSource) {
    ObjectMapper customMapper = JacksonMessagingUtils.messagingAwareMapper("com.example");
    customMapper.enable(SerializationFeature.INDENT_OUTPUT);
    customMapper.registerModule(new CustomModule());

    JacksonJsonObjectMapper jsonObjectMapper = new JacksonJsonObjectMapper(customMapper);

    JdbcChannelMessageStore store = new JdbcChannelMessageStore(dataSource);
    store.setPreparedStatementSetter(
        new JsonChannelMessageStorePreparedStatementSetter(jsonObjectMapper));
    store.setMessageRowMapper(
        new JsonMessageRowMapper(jsonObjectMapper));

    return store;
}

The custom JsonObjectMapper should be configured appropriately for Spring Integration message serialization. It is recommended to start with JacksonMessagingUtils.messagingAwareMapper() and customize from there. The same configuration must be used in both JsonChannelMessageStorePreparedStatementSetter and JsonMessageRowMapper for consistent serialization and deserialization.