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
JdbcAggregationRepositoryto 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?