JSON serialization for JDBC channel message store

[wimdeblauwe]

According to the documentation, you can use JSON as the serialization format for your JDBC channel message store by overriding the ChannelMessageStorePreparedStatementSetter. However, I found it was not that easy. If you only do that, the MessageRowMapper throws an exception that it cannot read what was written.

These are all the things I had to do to make it work:

1. Inject a custom queryProvider, preparedStatementSetter and messageRowMapper into the configuration of the JdbcChannelMessageStore:

  @Bean
  JdbcChannelMessageStore jdbcChannelMessageStore(
      DataSource dataSource,
      ChannelMessageStoreQueryProvider queryProvider,
      ChannelMessageStorePreparedStatementSetter preparedStatementSetter,
      MessageRowMapper messageRowMapper) {
    JdbcChannelMessageStore jdbcChannelMessageStore = new JdbcChannelMessageStore(dataSource);
    jdbcChannelMessageStore.setTablePrefix(CONCURRENT_METADATA_STORE_PREFIX);
    jdbcChannelMessageStore.setChannelMessageStoreQueryProvider(
        new PostgresChannelMessageStoreQueryProvider());
    jdbcChannelMessageStore.setChannelMessageStoreQueryProvider(queryProvider);
    jdbcChannelMessageStore.setPreparedStatementSetter(preparedStatementSetter);
    jdbcChannelMessageStore.setMessageRowMapper(messageRowMapper);
    return jdbcChannelMessageStore;
  }

2. Custom ChannelMessageStorePreparedStatementSetter that writes 2 colums: one with the type of the payload, and one with the serialized JSON of the payload.

  @Bean
  ChannelMessageStorePreparedStatementSetter channelMessageStorePreparedStatementSetter(ObjectMapper objectMapper) {
    return new ChannelMessageStorePreparedStatementSetter() {
      @Override
      public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage, Object groupId, String region,
          boolean priorityEnabled) throws SQLException {
        super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
        try {
          preparedStatement.setString(6, requestMessage.getPayload().getClass().getName());
          String json = objectMapper.writeValueAsString(requestMessage.getPayload());
          preparedStatement.setObject(7, json, java.sql.Types.OTHER);
        } catch (JsonProcessingException e) {
          throw new RuntimeException("Unable to store message", e);
        }
      }
    };
  }

3. Custom MessageRowMapper. Unfortunately, the setter at JdbcChannelMessageStore only accepts the MessageRowMapper class, not the RowMapper interface, but overriden the mapRow is luckily possible.

  @Bean
  MessageRowMapper messageRowMapper(ObjectMapper objectMapper) {
    return new MessageRowMapper(null, null) {
      @Override
      public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
        try {
          String payloadType = rs.getString(rs.findColumn("MESSAGE_PAYLOAD_TYPE"));
          String s = rs.getString(rs.findColumn("MESSAGE_BYTES"));
          Object o = objectMapper.readValue(s, Class.forName(payloadType));
          return new MutableMessage<>(o);
        } catch (JsonProcessingException | ClassNotFoundException e) {
          throw new RuntimeException("Unable to read message", e);
        }
      }
    };
  }

4. To support the extra column for the payload type, I needed to provide a custom query provider:

@Bean
  ChannelMessageStoreQueryProvider queryProvider() {
    return new PostgresChannelMessageStoreQueryProvider(){
      @Override
      public String getCreateMessageQuery() {
        return """
				INSERT into %PREFIX%CHANNEL_MESSAGE(
					MESSAGE_ID,
					GROUP_KEY,
					REGION,
					CREATED_DATE,
					MESSAGE_PRIORITY,
					MESSAGE_PAYLOAD_TYPE,
					MESSAGE_BYTES)
				values (?, ?, ?, ?, ?, ?, ?)
				""";
      }

      @Override
      public String getPollFromGroupQuery() {
        return """
				delete
				from %PREFIX%CHANNEL_MESSAGE
				where CTID = (select CTID
								from %PREFIX%CHANNEL_MESSAGE
								where %PREFIX%CHANNEL_MESSAGE.GROUP_KEY = :group_key
								and %PREFIX%CHANNEL_MESSAGE.REGION = :region
							order by CREATED_DATE, MESSAGE_SEQUENCE
							limit 1 for update skip locked)
				returning MESSAGE_ID, MESSAGE_PAYLOAD_TYPE, MESSAGE_BYTES;
				""";
      }

    };
  }

I also updated my Flyway script that creates the tables to add the MESSAGE_PAYLOAD_TYPE column:

CREATE TABLE _spring_integration_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('_spring_integration_MESSAGE_SEQ'),
    MESSAGE_PAYLOAD_TYPE VARCHAR      NOT NULL,
    MESSAGE_BYTES        JSON,
    REGION               VARCHAR(100) NOT NULL,
    CONSTRAINT _spring_integration_CHANNEL_MESSAGE_PK PRIMARY KEY (REGION, GROUP_KEY, CREATED_DATE, MESSAGE_SEQUENCE)
);

I wonder if this is the good approach to have (readable) JSON in the database instead of (unreadable) Java serialization in there? If so, maybe the documentation should be updated to provide a complete solution?

[artembilan]

Right. You made it so complex because you have to have that extra column for the payload type.
In other places we recommend to use JacksonJsonUtils.messagingAwareMapper() which adds this into the final JSON:

			init(JsonTypeInfo.Id.CLASS, null)
					.inclusion(JsonTypeInfo.As.PROPERTY);

So, the type info is stored together with data.
And when it is deserialized such an info is taken into account.

You suggestion about documenting this might make sense, but let's if solution having both type and data in the same JSON is more appropriate!

[wimdeblauwe]

I just tried it like this:

@Bean
  ChannelMessageStorePreparedStatementSetter channelMessageStorePreparedStatementSetter() {
    return new ChannelMessageStorePreparedStatementSetter() {
      @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 = JacksonJsonUtils.messagingAwareMapper().writeValueAsString(requestMessage.getPayload());
          preparedStatement.setObject(6, json, java.sql.Types.OTHER);
        } catch (JsonProcessingException e) {
          throw new RuntimeException("Unable to store message", e);
        }
      }
    };
  }

The resulting JSON did not contain class information:

{
  "subject":"Order 4 completed",
  "body":"Your order is registered in our system and will be processed.",
  "to":"[email protected]"
}

I then tried serializing the whole message instead of only the payload:

@Bean
  ChannelMessageStorePreparedStatementSetter channelMessageStorePreparedStatementSetter() {
    return new ChannelMessageStorePreparedStatementSetter() {
      @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 = JacksonJsonUtils.messagingAwareMapper().writeValueAsString(requestMessage);
          preparedStatement.setObject(6, json, java.sql.Types.OTHER);
        } catch (JsonProcessingException e) {
          throw new RuntimeException("Unable to store message", e);
        }
      }
    };
  }

The JSON then looks like this:

{
  "@class":"org.springframework.messaging.support.GenericMessage",
  "payload":{
    "@class":"com.wimdeblauwe.examples.transactional_outbox_spring_integration.infrastructure.mail.MailMessage",
    "subject":"Order 3 completed",
    "body":"Your order is registered in our system and will be processed.",
    "to":"[email protected]"
  },
  "headers":{
    "@class":"java.util.HashMap",
    "replyChannel":"nullChannel",
    "errorChannel":"",
    "id":[
      "java.util.UUID",
      "613c5543-a65a-cbbf-3cd9-1f5cba7ef83a"
    ],
    "timestamp":[
      "java.lang.Long",
      1720501984557
    ]
  }
}

Now there is class information, but I don't know if it should be the intention to store the whole message and not just the payload?

In both cases, there is an error message due to the MessageRowMapper not being compatible:

2024-07-09T07:18:02.078+02:00 ERROR 17370 --- [transactional-outbox-spring-integration] [   scheduling-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:433)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:355)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:56)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:54)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:348)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:96)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:317)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: org.springframework.core.serializer.support.SerializationFailedException: Failed to deserialize payload. Is the byte array a result of corresponding serialization for DefaultDeserializer?
	at org.springframework.integration.support.converter.AllowListDeserializingConverter.convert(AllowListDeserializingConverter.java:151)
	at org.springframework.integration.jdbc.store.channel.MessageRowMapper.mapRow(MessageRowMapper.java:63)
	at org.springframework.integration.jdbc.store.channel.MessageRowMapper.mapRow(MessageRowMapper.java:39)
	at org.springframework.jdbc.core.RowMapperResultSetExtractor.extractData(RowMapperResultSetExtractor.java:94)
	at org.springframework.jdbc.core.RowMapperResultSetExtractor.extractData(RowMapperResultSetExtractor.java:61)
	at org.springframework.jdbc.core.JdbcTemplate$1.doInPreparedStatement(JdbcTemplate.java:733)
	at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:658)
	at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:723)
	at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:748)
	at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:804)
	at org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate.query(NamedParameterJdbcTemplate.java:218)
	at org.springframework.integration.jdbc.store.JdbcChannelMessageStore.doPollForMessage(JdbcChannelMessageStore.java:608)
	at org.springframework.integration.jdbc.store.JdbcChannelMessageStore.pollMessageFromGroup(JdbcChannelMessageStore.java:557)
       .....
Caused by: java.io.StreamCorruptedException: invalid stream header: 7B224063
	at java.base/java.io.ObjectInputStream.readStreamHeader(ObjectInputStream.java:989)
	at java.base/java.io.ObjectInputStream.<init>(ObjectInputStream.java:416)
	at org.springframework.core.ConfigurableObjectInputStream.<init>(ConfigurableObjectInputStream.java:65)
	at org.springframework.core.ConfigurableObjectInputStream.<init>(ConfigurableObjectInputStream.java:51)
	at org.springframework.integration.support.converter.AllowListDeserializingConverter$1.<init>(AllowListDeserializingConverter.java:158)
	at org.springframework.integration.support.converter.AllowListDeserializingConverter.deserialize(AllowListDeserializingConverter.java:157)
	at org.springframework.integration.support.converter.AllowListDeserializingConverter.convert(AllowListDeserializingConverter.java:136)

So should the MessageRowMapper also need to be configured for JSON?

[wimdeblauwe]

Ok, figured it out completely. I did not use an AllowListDeserializingConverter since that uses byte[] and I want to user String. Anyway, I blogged how to configure it completely at Spring Integration using JSON serialization.

Thanks for the help!

[artembilan]

That's great!

I believe you can just rename the column into MESSAGE_CONTENT since it is not byte[] any more.

Also, take a look here where I provide the mentioned in your blog post Outbox pattern implementation: https://github.com/artembilan/microservices-patterns-spring-integration/tree/main/outbox

[artembilan]

I think we can contribute your components back to the framework to let others to benefit from your experience.
It really might so standard use-case when people want to have message serialized into JSON in the DB.

I'm raising an issue for future consideration.
Feel free to contribute the fix: https://github.com/spring-projects/spring-integration/blob/main/CONTRIBUTING.adoc

[artembilan]

Here you are: #9312 !

[wimdeblauwe]

Thanks!