How to migrate serialized exchange of a JdbcAggregationRepository?

Viewed 104

I have an Apache Camel application, which uses an Aggregator with a persistent aggregation repository. Sometimes my message model changes and I have to handle it.

Unfortunately the exchange isn't saved as JSON, XML or text, it is saved as a binary in a BLOB, see DATABASE:

DATABASE

[...]

The table structure definition of both table are identical: in both case a String value is used as key (id) whereas a Blob contains the exchange serialized in byte array.

Code

My model class is generated from JSON schema and implements Serializable to save the exchange in database:

public class MyMessage implements Serializable {

    @JsonProperty("value")
    private Integer value;
}

My route unmarshals a JSON message into my model class and aggregate it:

from(direct("input"))
    .unmarshal()
    .json(JsonLibrary.Jackson, MyMessage.class)
    .aggregate(header("CORRELATION_KEY"), myAggregationStrategy)
    .aggregationRepository(myAggregationRepository)
    .completionSize(2)
    .marshal()
    .json()
    .to("output");

My aggregation repository:

@Bean
public AggregationRepository myAggregationRepository(
    final PlatformTransactionManager platformTransactionManager, final DataSource dataSource) {

  final JdbcAggregationRepository repository = new JdbcAggregationRepository();
  repository.setRepositoryName("my_aggregation");
  repository.setTransactionManager(platformTransactionManager);
  repository.setDataSource(dataSource);

  return repository;
}

I create the database tables with Flyway:

CREATE TABLE my_aggregation (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 version BIGINT NOT NULL,
 constraint my_aggregation_pk PRIMARY KEY (id)
);
CREATE TABLE my_aggregation_completed (
 id varchar(255) NOT NULL,
 exchange blob NOT NULL,
 version BIGINT NOT NULL,
 constraint my_aggregation_completed_pk PRIMARY KEY (id)
);

Research

Apache Camel supports to save message body as text, see STORING BODY AND HEADERS AS TEXT:

You can configure the JdbcAggregationRepository to store message body and select(ed) headers as String in separate columns.

But this additional column is not read by the aggregator.

Question

How can I migrate persisted exchanges in my database?

0 Answers
Related