This version is still in development and is not considered stable yet. For the latest stable version, please use Spring Integration 6.3.2! |
Debezium Support
Debezium Engine, Change Data Capture (CDC) inbound channel adapter.
The DebeziumMessageProducer
allows capturing database change events, converting them into messages and streaming later to the outbound channels.
You need to include the spring integration Debezium dependency to your project:
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-debezium</artifactId>
<version>6.3.3-SNAPSHOT</version>
</dependency>
compile "org.springframework.integration:spring-integration-debezium:6.3.3-SNAPSHOT"
You also need to include a debezium connector dependency for your input Database. For example to use Debezium with PostgreSQL you will need the postgres debezium connector:
-
Maven
-
Gradle
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-postgres</artifactId>
<version>${debezium-version}</version>
</dependency>
compile "io.debezium:debezium-connector-postgres:{debezium-version}"
Replace the |
Inbound Debezium Channel Adapter
The Debezium adapter expects a pre-configured DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>>
instance.
The debezium-supplier provides an out of the box |
The Debezium Java DSL can create a |
Additionally, the DebeziumMessageProducer
can be tuned with the following configuration properties:
-
contentType
- allows handling forJSON
(default),AVRO
andPROTOBUF
message contents. The contentTypemust
be be aligned with theSerializationFormat
configured for the providedDebeziumEngine.Builder
. -
enableBatch
- when set tofalse
(default), the debezium adapter would send newMessage
for everyChangeEvent
data change event received from the source database. If set totrue
then the adapter sends downstream a singleMessage
for each batch ofChangeEvent
received from the Debezium engine. Such a payload is not serializable and would require a custom serialization/deserialization implementation. -
enableEmptyPayload
- Enables support for tombstone (aka delete) messages. On a database row delete, Debezium can send a tombstone change event that has the same key as the deleted row and a value ofOptional.empty
. Defaults tofalse
. -
headerMapper
- customHeaderMapper
implementation that allows for selecting and converting theChangeEvent
headers intoMessage
headers. The defaultDefaultDebeziumHeaderMapper
implementation provides a setter forsetHeaderNamesToMap
. By default, all headers are mapped. -
taskExecutor
- Set a customTaskExecutor
for the Debezium engine.
The following code snippets demonstrate various configuration for this channel adapter:
Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:
@SpringBootApplication
public class DebeziumJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(DebeziumJavaApplication.class)
.web(WebApplicationType.NONE)
.run(args);
}
@Bean
public MessageChannel debeziumInputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducer debeziumMessageProducer(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder,
MessageChannel debeziumInputChannel) {
DebeziumMessageProducer debeziumMessageProducer =
new DebeziumMessageProducer(debeziumEngineBuilder);
debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
return debeziumMessageProducer;
}
@ServiceActivator(inputChannel = "debeziumInputChannel")
public void handler(Message<?> message) {
Object destination = message.getHeaders().get(DebeziumHeaders.DESTINATION); (1)
String key = new String((byte[]) message.getHeaders().get(DebeziumHeaders.KEY)); (2)
String payload = new String((byte[]) message.getPayload()); (3)
System.out.println("KEY: " + key + ", DESTINATION: " + destination + ", PAYLOAD: " + payload);
}
}
1 | A name of the logical destination for which the event is intended.
Usually the destination is composed of the topic.prefix configuration option, the database name and the table name. For example: my-topic.inventory.orders . |
2 | Contains the schema for the changed table’s key and the changed row’s actual key.
Both the key schema and its corresponding key payload contain a field for each column in the changed table’s PRIMARY KEY (or unique constraint) at the time the connector created the event. |
3 | Like the key, the payload has a schema section and a payload value section. The schema section contains the schema that describes the Envelope structure of the payload value section, including its nested fields. Change events for operations that create, update or delete data all have a value payload with an envelope structure. |
The |
Similarly, we can configure the DebeziumMessageProducer
to process the incoming change events in batches:
@Bean
public MessageProducer debeziumMessageProducer(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder,
MessageChannel debeziumInputChannel) {
DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder);
debeziumMessageProducer.setEnableBatch(true);
debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
return debeziumMessageProducer;
}
@ServiceActivator(inputChannel = "debeziumInputChannel")
public void handler(List<ChangeEvent<Object, Object>> payload) {
System.out.println(payload);
}
Debezium Java DSL Support
The spring-integration-debezium
provides a convenient Java DSL fluent API via the Debezium
factory and the DebeziumMessageProducerSpec
implementations.
The Inbound Channel Adapter for Debezium Java DSL is:
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder = ...
IntegrationFlow.from(
Debezium.inboundChannelAdapter(debeziumEngineBuilder)
.headerNames("special*")
.contentType("application/json")
.enableBatch(false))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))
Or create an DebeziumMessageProducerSpec
instance from native debezium configuration properties and default to JSON
serialization formats.
Properties debeziumConfig = ...
IntegrationFlow
.from(Debezium.inboundChannelAdapter(debeziumConfig))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))
The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL:
@SpringBootApplication
public class DebeziumJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(DebeziumJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow debeziumInbound(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
return IntegrationFlow
.from(Debezium
.inboundChannelAdapter(debeziumEngineBuilder)
.headerNames("special*")
.contentType("application/json")
.enableBatch(false))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))
.get();
}
}