[[debezium]]
= Debezium Support

https://debezium.io/documentation/reference/development/engine.html[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:

[tabs]
======
Maven::
+
[source, xml, subs="normal", role="primary"]
----
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-debezium</artifactId>
    <version>{project-version}</version>
</dependency>
----

Gradle::
+
[source, groovy, subs="normal", role="secondary"]
----
compile "org.springframework.integration:spring-integration-debezium:{project-version}"
----
======

You also need to include a https://debezium.io/documentation/reference/connectors/index.html[debezium connector] dependency for your input Database.
For example to use Debezium with PostgreSQL you will need the postgres debezium connector:

[tabs]
======
Maven::
+
[source, xml, role="primary"]
----
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-postgres</artifactId>
    <version>${debezium-version}</version>
</dependency>

----

Gradle::
+
[source, groovy, role="secondary"]
----
compile "io.debezium:debezium-connector-postgres:{debezium-version}"
----
======

[NOTE]
====
Replace the `debezium-version` with the version compatible with the `spring-integration-debezium` version being used.
====

[[debezium-inbound]]
== Inbound Debezium Channel Adapter

The Debezium adapter expects a pre-configured `DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>>` instance.

[TIP]
====
The https://github.com/spring-cloud/stream-applications/tree/main/functions/supplier/debezium-supplier[debezium-supplier] provides an out of the box `DebeziumEngine.Builder` Spring Boot auto-configuration with a handy https://github.com/spring-cloud/stream-applications/blob/main/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java[DebeziumProperties] configuration abstraction.
====

[TIP]
====
The xref:debezium.adoc#debezium-java-dsl[Debezium Java DSL] can create a `DebeziumMessageProducer` instance from a provided `DebeziumEngine.Builder`, as well as from a plain Debezium configuration (e.g. `java.util.Properties`).
Later can be handy for some common use-cases with opinionated configuration and serialization formats.
====

Additionally, the `DebeziumMessageProducer` can be tuned with the following configuration properties:

- `contentType` - allows handling for  `JSON` (default), `AVRO` and `PROTOBUF` message contents.
The contentType `must` be be aligned with the `SerializationFormat` configured for the provided `DebeziumEngine.Builder`.
- `enableBatch` - when set to `false` (default), the debezium adapter would send new `Message` for every `ChangeEvent` data change event received from the source database.
If set to `true` then the adapter sends downstream a single `Message` for each batch of `ChangeEvent` 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 of `Optional.empty`.
Defaults to `false`.
- `headerMapper` - custom `HeaderMapper` implementation that allows for selecting and converting the `ChangeEvent` headers into `Message` headers.
The default `DefaultDebeziumHeaderMapper` implementation provides a setter for `setHeaderNamesToMap`.
By default, all headers are mapped.
- `taskExecutor` - Set a custom `TaskExecutor` for the Debezium engine.

The following code snippets demonstrate various configuration for this channel adapter:

[[configuring-with-java-configuration]]
=== Configuring with Java Configuration

The following Spring Boot application shows an example of how to configure the inbound adapter with Java configuration:

[source, java]
----
@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.

[TIP]
====
The `key.converter.schemas.enable=false` and/or `value.converter.schemas.enable=false` permit disabling the in-message schema content for key or payload respectively.
====

Similarly, we can configure the `DebeziumMessageProducer` to process the incoming change events in batches:

[source, java]
----
@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]]
== 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:
[source, java]
----
 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.

[source, java]
----
 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:

[source, java]
----
@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();
    }

}
----
