SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move Debezium builder into standalone function common project (#463)

Browse Source 
- Move the `DebeziumEngineBuilderAutoConfiguration` and META-INF configurations to new function project: debezium-autoconfigure.
 - Introduce the `EmbeddedEngineExecutorService` helper class for help DebezuimEngine implementations.
 - Move the relevant tests from debezium-supplier to the debeizum-autoconfigure.
 - Add detailed README documentation.
 - Adjust the debeizum-supplier and debezium-source configurations.
 - Clean the related tests.

 Resolves #462


---------

Co-authored-by: Artem Bilan <abilan@vmware.com>
This commit is contained in:
Christian Tzolov
2023-06-04 17:58:12 +02:00
committed by GitHub
parent 898938dd55
commit 11be59afd1
25 changed files with 636 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
applications/source/debezium-source/README.adoc
View File

@@ -18,12 +18,20 @@ The `Debezium Source` currently supports CDC for multiple datastores: https://de
== Options
//tag::configuration-properties[]
$$debezium.copy-headers$$:: $$Copy Change Event headers into Message headers.$$ *($$Boolean$$, default: `$$true$$`)*
$$debezium.debezium-native-configuration$$:: $$<documentation missing>$$ *($$Properties$$, default: `$$<none>$$`)*
$$debezium.header-format$$:: $${@link ChangeEvent} header format. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$<none>$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)*
$$debezium.offset-commit-policy$$:: $$The policy that defines when the offsets should be committed to offset storage.$$ *($$DebeziumOffsetCommitPolicy$$, default: `$$<none>$$`, possible values: `ALWAYS`,`PERIODIC`,`DEFAULT`)*
$$debezium.payload-format$$:: $${@link ChangeEvent} Key and Payload formats. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$<none>$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)*
$$debezium.properties$$:: $$Spring pass-trough wrapper for debezium configuration properties. All properties with a 'debezium.properties.*' prefix are native Debezium properties.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
Properties grouped by prefix:
=== debezium
$$debezium-native-configuration$$:: $$<documentation missing>$$ *($$Properties$$, default: `$$<none>$$`)*
$$header-format$$:: $${@link ChangeEvent} header format. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$<none>$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)*
$$offset-commit-policy$$:: $$The policy that defines when the offsets should be committed to offset storage.$$ *($$DebeziumOffsetCommitPolicy$$, default: `$$<none>$$`, possible values: `ALWAYS`,`PERIODIC`,`DEFAULT`)*
$$payload-format$$:: $${@link ChangeEvent} Key and Payload formats. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$<none>$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)*
$$properties$$:: $$Spring pass-trough wrapper for debezium configuration properties. All properties with a 'debezium.properties.*' prefix are native Debezium properties.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
=== debezium.supplier
$$copy-headers$$:: $$Copy Change Event headers into Message headers.$$ *($$Boolean$$, default: `$$true$$`)*
//end::configuration-properties[]
==== Event flattening configuration
@@ -156,10 +164,10 @@ Use following properties to connect the Debezium Source to MySQL DB:
----
debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector # <1>
debezium.properties.topic.prefix=my-topic # <2>
debezium.properties.name=my-connector # <2>
debezium.properties.topic.prefix=my-topic # <2>
debezium.properties.database.server.id=85744 # <2>
debezium.properties.database.server.name=my-app-connector # <2>
debezium.properties.database.user=debezium # <3>
debezium.properties.database.password=dbz # <3>
@@ -176,7 +184,6 @@ debezium.properties.transforms.unwrap.add.fields=name,db # <5>
debezium.properties.transforms.unwrap.delete.handling.mode=none # <5>
debezium.properties.transforms.unwrap.drop.tombstones=true # <5>
debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory # <6>
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory # <6>
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore # <6>
@@ -214,14 +221,12 @@ Use following properties to connect the Debezium Source to PostgreSQL:
----
debezium.properties.connector.class=io.debezium.connector.postgresql.PostgresConnector # <1>
debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory # <2>
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory # <2>
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore # <2>
debezium.properties.topic.prefix=my-topic # <3>
debezium.properties.name=my-connector # <3>
debezium.properties.database.server.id=85744 # <3>
debezium.properties.database.server.name=my-app-connector # <3>
debezium.properties.database.user=postgres # <4>
debezium.properties.database.password=postgres # <4>
@@ -282,14 +287,12 @@ debezium.properties.connector.class=io.debezium.connector.mongodb.MongodbSourceC
debezium.properties.topic.prefix=my-topic
debezium.properties.name=my-connector
debezium.properties.database.server.id=85744
debezium.properties.database.server.name=my-app-connector
debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory # <2>
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory # <2>
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore # <2>
debezium.properties.mongodb.hosts=rs0/localhost:27017 # <3>
debezium.properties.mongodb.name=dbserver1 # <3>
debezium.properties.topic.prefix=dbserver1 # <3>
debezium.properties.mongodb.user=debezium # <3>
debezium.properties.mongodb.password=dbz # <3>
debezium.properties.database.whitelist=inventory # <3>
@@ -337,14 +340,12 @@ Use following properties to connect the Debezium Source to SQLServer:
----
debezium.properties.connector.class=io.debezium.connector.sqlserver.SqlServerConnector # <1>
debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory # <2>
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory # <2>
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore # <2>
debezium.properties.topic.prefix=my-topic # <3>
debezium.properties.name=my-connector # <3>
debezium.properties.database.server.id=85744 # <3>
debezium.properties.database.server.name=my-app-connector # <3>
debezium.properties.database.user=sa # <4>
debezium.properties.database.password=Password! # <4>
@@ -379,7 +380,7 @@ cat ./inventory.sql | docker exec -i dbz_oracle sqlplus debezium/dbz@//localhost
[source,shell]
----
java -jar debezium-source.jar --debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector --debezium.properties.topic.prefix=my-topic --debezium.properties.name=my-connector --debezium.properties.database.server.id=85744 --debezium.properties.database.server.name=my-app-connector --debezium.properties.database.server.id=85744 --debezium.properties.database.server.name=my-app-connector --debezium.properties.database.user=debezium --debezium.properties.database.password=dbz --debezium.properties.database.hostname=localhost --debezium.properties.database.port=3306 --debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory --debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory --debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore
java -jar debezium-source.jar --debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector --debezium.properties.topic.prefix=my-topic --debezium.properties.name=my-connector --debezium.properties.database.server.id=85744 --debezium.properties.database.server.id=85744 --debezium.properties.database.user=debezium --debezium.properties.database.password=dbz --debezium.properties.database.hostname=localhost --debezium.properties.database.port=3306 --debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory --debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore
----

1
applications/source/debezium-source/pom.xml
View File

@@ -12,7 +12,6 @@
<properties>
<json-unit.version>1.25.1</json-unit.version>
<!-- <mysql-connector-java.version>8.0.13</mysql-connector-java.version> -->
</properties>
<artifactId>debezium-source</artifactId>

2
applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties
View File

@@ -1 +1 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.debezium.DebeziumProperties
configuration-properties.classes=org.springframework.cloud.fn.common.debezium.DebeziumProperties,org.springframework.cloud.fn.supplier.debezium.DebeziumSupplierProperties

2
applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties
View File

@@ -1 +1 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.debezium.DebeziumProperties
configuration-properties.classes=org.springframework.cloud.fn.common.debezium.DebeziumProperties,org.springframework.cloud.fn.supplier.debezium.DebeziumSupplierProperties

8
applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/databases/DebeziumDatabasesIntegrationTest.java
View File

@@ -82,8 +82,7 @@ public class DebeziumDatabasesIntegrationTest {
"debezium.properties.topic.prefix=my-topic",
"debezium.properties.name=my-connector",
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector");
"debezium.properties.database.server.id=85744");
@Test
public void mysql() {
@@ -227,10 +226,7 @@ public class DebeziumDatabasesIntegrationTest {
"--debezium.properties.tasks.max=1",
"--debezium.properties.mongodb.connection.string=mongodb://" + host + ":" + port
+ "/?replicaSet=rs0",
// "--debezium.properties.mongodb.connection.string=mongodb://" + host + ":"
// + "27017"
// + "/?replicaSet=rs0",
"--debezium.properties.mongodb.name=dbserver1",
"--debezium.properties.topic.prefix=dbserver1",
"--debezium.properties.mongodb.user=debezium",
"--debezium.properties.mongodb.password=dbz",
"--debezium.properties.collection.include.list=inventory[.]*")) {

8
applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumDeleteHandlingIntegrationTest.java
View File

@@ -29,7 +29,7 @@ import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.boot.test.context.runner.ContextConsumer;
import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
@@ -70,14 +70,14 @@ public class DebeziumDeleteHandlingIntegrationTest {
"debezium.properties.key.converter.schemas.enable=false",
"debezium.properties.value.converter.schemas.enable=false",
"debezium.properties.topic.prefix=my-topic", // new
"debezium.properties.topic.prefix=my-topic",
// enable flattering
"debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.add.fields=name,db",
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory", // new
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory",
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"debezium.properties.name=my-connector",
@@ -87,8 +87,6 @@ public class DebeziumDeleteHandlingIntegrationTest {
"debezium.properties.database.hostname=localhost",
"debezium.properties.database.port=" + mySqlContainer.getMappedPort(3306),
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector",
"debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory",
// JdbcTemplate configuration
String.format("app.datasource.url=jdbc:mysql://localhost:%d/%s?enabledTLSProtocols=TLSv1.2",

8
applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumFlatteningIntegrationTest.java
View File

@@ -30,7 +30,7 @@ import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.boot.test.context.runner.ContextConsumer;
import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
@@ -73,11 +73,11 @@ public class DebeziumFlatteningIntegrationTest {
"debezium.properties.key.converter.schemas.enable=false",
"debezium.properties.value.converter.schemas.enable=false",
"debezium.properties.topic.prefix=my-topic", // new
"debezium.properties.topic.prefix=my-topic",
"debezium.properties.name=my-sql-connector",
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory", // new
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory",
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
@@ -86,8 +86,6 @@ public class DebeziumFlatteningIntegrationTest {
"debezium.properties.database.hostname=localhost",
"debezium.properties.database.port=" + mySqlContainer.getMappedPort(3306),
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector",
"debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory",
// JdbcTemplate configuration
String.format("app.datasource.url=jdbc:mysql://localhost:%d/%s?enabledTLSProtocols=TLSv1.2",

1
applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumSupplierAvroFormatTest.java
View File

@@ -79,7 +79,6 @@ public class DebeziumSupplierAvroFormatTest {
"debezium.properties.topic.prefix=my-topic",
"debezium.properties.name=my-connector",
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
"debezium.properties.database.user=debezium",

353
functions/common/debezium-autoconfigure/README.adoc Normal file
View File

@@ -0,0 +1,353 @@
= Debezium Auto-Configuration
This module provides a generic https://debezium.io/documentation/reference/2.2/development/engine.html[DebeziumEngine.Builder] auto-configuration that can be reused and composed in other applications.
IMPORTANT: The `DebeziumEngine` does not required Kafka or Kafka Connect as it runs embedded inside your application.
This approach though comes with some delivery guarantee limitations as explained https://debezium.io/documentation/reference/2.2/development/engine.html#%5Fhandling_failures[here].
The `Debezium Engine` is a https://en.wikipedia.org/wiki/Change_data_capture[Change Data Capture] (CDC) utility, that allows *capturing* database change events and process them with custom `java.util.Consumer` or `io.debezium.engine.ChangeConsumer` event handler implementations.
The `DebeziumEngine.Builder` auto-configuration is activated only if a https://debezium.io/documentation/reference/stable/connectors/index.html[Debezium Connector] is found on the classpath and the `debezium.properties.connector.class` property is set to point to that connector class.
== Quick Start
To use the `DebeziumEngine`, set the required <<dependencies>> and <<configuration-options>>, register a custom <<changeEvent-handler>> with the engine and submit later for execution.
Those simple steps are illustrated in the following sections.
[[dependencies]]
=== Dependencies
To process the incoming database change events you need to include the Debezium `auto-configuration` dependency to your project:
====
[source, xml, subs="normal", role="primary"]
.Maven
----
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>debezium-autoconfigure</artifactId>
<version>${project-version}</version>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
----
compile "org.springframework.cloud.fn:debezium-autoconfigure:{project-version}"
----
====
and include the https://debezium.io/documentation/reference/2.2/connectors/index.html[debezium connector] dependency for the selected Database.
For example the postgres debezium connector dependency looks like this:
====
[source, xml, subs="normal", role="primary"]
.Maven
----
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-postgres</artifactId>
<version>${debezium-version}</version>
</dependency>
----
[source, groovy, subs="normal", role="secondary"]
.Gradle
----
compile "io.debezium:debezium-connector-postgres:{debezium-version}"
----
====
[[changeEvent-handler]]
### ChangeEvent Handler
To process the incoming change events, implement a `java.util.Consumer<ChangeEvent>` (or `ChangeConsumer<ChangeEvent>`) handler and wire it into the `DebeziumEngine.Builder`.
Then build an engine from the builder and run it from within an Executor service:
[source, java]
----
public void simple(DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> builder) {
Executors.newSingleThreadExecutor().execute(builder // <3>
.notifying(changeEvent -> System.out.println(changeEvent.value())) // <1>
.build()); // <2>
}
----
<1> Implement and register a change events consumer.
<2> Build the Debezium engine.
<3> Submit the engine to a single thread executor.
For a real life applications you should manage the lifecycles of the engine and its executor.
The Spring `Lifecycle` and `SmartLifecycle` interfaces can help you to manage this.
Here is a more realistic example snipped:
[source, java]
----
@Bean
public Consumer<ChangeEvent<byte[], byte[]>> customConsumer() { // <1>
return new Consumer<ChangeEvent<byte[], byte[]>>() {
@Override
public void accept(ChangeEvent<byte[], byte[]> changeEvent) {
if (changeEvent != null) { // ignore null records
System.out.println("Key:" + changeEvent.key() + ", Value: " changeEvent.value());
}
}
}
}
@Bean
public DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine( // <2>
Consumer<ChangeEvent<byte[], byte[]>> consumer,
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> builder) {
return new builder.notifying(consumer).build();
}
@Bean
public EmbeddedEngineExecutorService embeddedEngine( // <3>
DebeziumEngine<ChangeEvent<byte[], byte[]>> debeziumEngine) {
return new EmbeddedEngineExecutorService(debeziumEngine);
}
----
<1> Create a custom change event consumer.
<2> Use the consumer from step (1) and the debezium builder provided from the auto-configuration to create a new DebeziumEngine instance.
<3> The DebeziumEngine is designed to be submitted to an `Executor` or `ExecutorService` for execution.
The `EmbeddedEngineExecutorService` is a convenient `ExecutorService` implementation, aligned with the Spring lifecycle.
NOTE: The `EmbeddedEngineExecutorService` is optional.
Some application such as `DebeziumReactiveConsumerConfiguration` implement the ExecutorService as part of their
`Supplier<Flux<Message<?>>>` configuration.
NOTE: Neither the `Consumer`/`ChangeConsumer` nor the `DebeziumEngine` need to be `@Bean`!
It is enough to set some consumer into it the `DebeziumEngine.Builder`, build an engine and run it from within some `@Service` implementation.
[[configuration-options]]
== Configuration Options
$$debezium.header-format$$:: `ChangeEvent` header format.
*(default and only option at the moment: `JSON`)*
$$debezium.payload-format$$:: `ChangeEvent` Key and Payload formats.
*($$DebeziumFormat$$, default: `JSON`, possible values: `JSON`,`AVRO`,`PROTOBUF`)*
$$debezium.offset-commit-policy$$:: The policy that defines when the offsets should be committed to offset storage.
*($$DebeziumOffsetCommitPolicy$$, default: `PERIODIC`, possible values: `ALWAYS`,`PERIODIC`,`DEFAULT`)*
$$debezium.properties$$:: $$Spring pass-trough wrapper for debezium configuration properties.
All properties with a `debezium.properties.*` prefix are native Debezium properties.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*.
For example the `debezium.properties.connector.class` property is converted into `connector.class` before provided to the DebeziumEngine.
Here is an example configuration for the sample snipped above:
[source, bash]
----
debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector # <1>
debezium.properties.database.user=debezium # <2>
debezium.properties.database.password=dbz # <2>
debezium.properties.database.hostname=localhost # <2>
debezium.properties.database.port=3306 # <2>
debezium.properties.name=my-sql-connector # <3>
debezium.properties.database.server.id=85744 # <3>
debezium.properties.topic.prefix=my-topic # <3>
debezium.properties.key.converter.schemas.enable=true # <4>
debezium.properties.value.converter.schemas.enable=true # <4>
debezium.properties.offset.flush.interval.ms=60000
debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory # <5>
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore # <5>
debezium.header-format=JSON # <6>
debezium.payload-format=JSON # <6>
----
<1> Configures the Debezium Engine to use https://debezium.io/docs/connectors/mysql/[MySqlConnector].
<2> Configure the connection to a MySQL server running on `localhost:3306` as `debezium` user.
<3> Metadata used to identify and dispatch the incoming events.
* `debezium.properties.topic.prefix` - provides a namespace for the particular database server/cluster in which Debezium is capturing changes.
The topic prefix **should be unique** across all other connectors.
Only alphanumeric characters, hyphens, dots and underscores must be used.
* `debezium.properties.database.server.id` - a numeric identifier of this database client, which **must be unique across all currently-running database processes**.
<4> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `ChangeEvent` message.
<5> Metadata stores to preserver the debezium state between multiple starts.
<6> Sets, explicitly, the ChangeEvent header and payload (e.g. key and value) serialization formats.
Defaults to JSON with binary encoding.
=== Connectors properties
The table below lists all available Debezium properties for each connecter.
.Table of the native Debezium configuration properties for every connector.
|===
| Connector | Connector properties
|https://debezium.io/documentation/reference/2.2/connectors/mysql.html[MySQL]
|https://debezium.io/documentation/reference/2.2/connectors/mysql.html#mysql-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/mongodb.html[MongoDB]
|https://debezium.io/documentation/reference/2.2/connectors/mongodb.html#mongodb-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/postgresql.html[PostgreSQL]
|https://debezium.io/documentation/reference/2.2/connectors/postgresql.html#postgresql-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/oracle.html[Oracle]
|https://debezium.io/documentation/reference/2.2/connectors/oracle.html#oracle-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html[SQL Server]
|https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html#sqlserver-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/db2.html[DB2]
|https://debezium.io/documentation/reference/2.2/connectors/db2.html#db2-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/vitess.html[Vitess]
|https://debezium.io/documentation/reference/2.2/connectors/vitess.html#vitess-connector-properties
|https://debezium.io/documentation/reference/2.2/connectors/spanner.html[Spanner]
|https://debezium.io/documentation/reference/2.2/connectors/spanner.html#spanner-connector-properties
|===
=== Streaming vs Batching
If you register a `java.util.Consumer<ChangeEvent>` with the `DebeziumEngine.Builder` then the incoming events are processed element-wise, one by one in the order of their occurrence in the source database.
Opting for the `io.debezium.engineChangeConsumer<ChangeEvent>` provides an https://debezium.io/documentation/reference/stable/development/engine.html#advanced-consuming[advanced event consumption] that can process batch of events in one go, acknowledging their processing once that's done.
This snippet illustrates how to implement a batch handler:
[source, java]
----
@Bean
public EmbeddedEngineExecutorService batch(DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> builder) {
return new EmbeddedEngineExecutorService( // <3>
builder.notifying(new ChangeConsumer<>() { // <1>
@Override
public void handleBatch(
List<ChangeEvent<byte[], byte[]>> changeEventBatch,
RecordCommitter<ChangeEvent<byte[], byte[]>> committer)
throws InterruptedException {
for (ChangeEvent<byte[], byte[]> event : changeEventBatch) {
System.out.println(event.value());
committer.markProcessed(event);
}
committer.markBatchFinished();
}
}).build()); // <2>
}
----
<1> Implement and register a `ChangeConsumer` batch handler.
The `committer.markProcessed(event)` and `committer.markBatchFinished()` are used to mark the event and batch completion.
<2> Build the engine.
<3> Crate and return an `EmbeddedEngineExecutorService` - a Spring lifecycle manageable `ExecutorService`.
=== Additional Configuration Components
The Debezium builder auto-configuration provides an opinionated implementation for the following configurable components:
- `OffsetCommitPolicy` - Commit policy type.
The default is a periodic commit policy based upon time intervals.
- `Clock` - Clock needing to determine the current time.
Defaults to the `Clock#systemDefaultZone()` system clock.
- `CompletionCallback` - callback called by the engine on `DebeziumEngine#run()` method completes with the results.
By default logs the completion status.
- `ConnectorCallback` - During the engine run, provides feedback about the the completion state of each component running within the engine (connectors, tasks etc).
By default logs the connector state.
You can override any of the above components.
Just provide your `@Bean` implementation to the application context.
=== Event Flattening
Debezium provides a comprehensive message format, that accurately details information about changes that happen in the system.
Sometime this format, though, might not be suitable for the downstream consumers, that might require messages that are formatted so that field names and values are presented in a simplified, `flattened` structure.
To simplify the format of the event records that the Debezium connectors produce, you can use the https://debezium.io/documentation/reference/stable/transformations/event-flattening.html[Debezium event flattening] message transformation:
[source, bash]
----
debezium.properties.transforms=flattening # <1>
debezium.properties.transforms.flattening.type=io.debezium.transforms.ExtractNewRecordState # <2>
debezium.properties.transforms.flattening.drop.tombstones=false # <3>
debezium.properties.transforms.flattening.delete.handling.mode=rewrite # <4>
debezium.properties.transforms.flattening.add.headers=op # <5>
debezium.properties.transforms.flattening.add.fields=name,db # <5>
----
<1> flattening transformation name.
<2> Class that implements the flatting transformation.
<3> Debezium generates a tombstone record for each DELETE operation.
The default behavior is that event flattening removes tombstone records from the stream.
To keep tombstone records in the stream, specify drop.tombstones=false.
<4> Debezium generates a change event event for each DELETE operation.
The `rewrite` mode keeps those events, which a dropped otherwise.
<5> Comma-separated list of metadata fields to add to the header and the value of the simplified event value.
Follow the https://debezium.io/documentation/reference/stable/transformations/event-flattening.html#_configuration[configuration documentation] for further details.
=== Offset Storages
When a Debezium source runs, it reads information from the source and periodically records `offsets` that define how much of that information it has processed.
Should the source be restarted, it will use the last recorded offset to know where in the source information it should resume reading.
Out of the box, the following https://debezium.io/documentation/reference/2.2/development/engine.html#engine-properties[offset storage configuration] options are provided:
==== In-Memory
Doesn't persist the offset data but keeps it in memory.
Therefore all offsets are lost on debezium source restart.
=====
[source, bash]
----
debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore
----
=====
==== Local Filesystem
Store the offsets in a file on the local file system (the file can be named anything and stored anywhere).
Additionally, although the connector records the offsets with every source record it produces, the engine flushes the offsets to the backing store periodically (in the example below, once each minute).
=====
[source, bash]
----
debezium.properties.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
debezium.properties.offset.storage.file.filename=/tmp/offsets.dat # <1>
debezium.properties.offset.flush.interval.ms=60000 # <2>
----
<1> Path to file where offsets are to be stored.
Required when `offset.storage`` is set to the `FileOffsetBackingStore`.
<2> Interval at which to try committing offsets.
The default is 1 minute.
=====
==== Kafka topic
Uses a Kafka topic to store offset data.
=====
[source, bash]
----
debezium.properties.offset.storage=org.apache.kafka.connect.storage.KafkaOffsetBackingStore
debezium.properties.offset.storage.topic=my-kafka-offset-topic # <1>
debezium.properties.offset.storage.partitions=2 # <2>
debezium.properties.offset.storage.replication.factor=1 # <3>
debezium.properties.offset.flush.interval.ms=60000 # <4>
----
<1> The name of the Kafka topic where offsets are to be stored.
Required when `offset.storage` is set to the `KafkaOffsetBackingStore`.
<2> The number of partitions used when creating the offset storage topic.
<3> Replication factor used when creating the offset storage topic.
<4> Interval at which to try committing offsets.
The default is 1 minute.
=====
One can implement the `org.apache.kafka.connect.storage.OffsetBackingStore` interface in to provide a offset storage bound to a custom backend key-value store.
== Tests
See this link:org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java[test suite] for how to use the auto-configuration with custom Consumer.
== Other usage
- See the https://github.com/spring-cloud/stream-applications/blob/master/functions/supplier/debezium-source/debezium-supplier[debezium-supplier] implementation about how to implement reactive consumer on top of the debezium auto-configuration.
- See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/debezium-source/README.adoc[debezium-source] about how the debezium auto-configuration and supplier are used to create a Spring Cloud Stream applications.
- See the https://docs.spring.io/spring-integration/docs/6.2.0-SNAPSHOT/reference/html/debezium.html#debezium[Spring Integration Debezium support] about how to initialize Inbound Debezium Channel Adapter with `DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>>` provided by the auto-configuration.

88
functions/common/debezium-autoconfigure/pom.xml Normal file
View File

@@ -0,0 +1,88 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
</parent>
<artifactId>debezium-autoconfigure</artifactId>
<name>debezium-autoconfigure</name>
<description>Debezium Spring Boot auto-configuration</description>
<properties>
<version.debezium>2.2.1.Final</version.debezium>
<apicurio.version>2.4.2.Final</apicurio.version>
</properties>
<dependencies>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-embedded</artifactId>
<version>${version.debezium}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-reload4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- APICURIO Schema Registry (for AVRO format)-->
<dependency>
<groupId>io.apicurio</groupId>
<artifactId>apicurio-registry-utils-converter</artifactId>
<version>${apicurio.version}</version>
</dependency>
<dependency>
<groupId>io.apicurio</groupId>
<artifactId>apicurio-registry-client</artifactId>
<version>${apicurio.version}</version>
</dependency>
<!-- TEST -->
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mysql</artifactId>
<version>${version.debezium}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>4.0.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.0.0</version>
<configuration>
<groups>integration</groups>
</configuration>
</plugin>
</plugins>
</build>
</project>

6
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java → functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfiguration.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium;
package org.springframework.cloud.fn.common.debezium;
import java.time.Clock;
import java.time.Duration;
@@ -38,7 +38,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties.DebeziumFormat;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.core.Ordered;
@@ -123,7 +123,7 @@ public class DebeziumEngineBuilderAutoConfiguration {
}
@Bean
public Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder(
public DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder(
OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback,
ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) {

15
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java → functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumProperties.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium;
package org.springframework.cloud.fn.common.debezium;
import java.util.HashMap;
import java.util.Map;
@@ -69,11 +69,6 @@ public class DebeziumProperties {
*/
private DebeziumFormat headerFormat = DebeziumFormat.JSON;
/**
* Copy Change Event headers into Message headers.
*/
private boolean copyHeaders = true;
/**
* The policy that defines when the offsets should be committed to offset storage.
*/
@@ -99,14 +94,6 @@ public class DebeziumProperties {
this.headerFormat = headerFormat;
}
public boolean isCopyHeaders() {
return copyHeaders;
}
public void setCopyHeaders(boolean copyHeaders) {
this.copyHeaders = copyHeaders;
}
public enum DebeziumOffsetCommitPolicy {
/**
* Commits offsets as frequently as possible. This may result in reduced performance, but it has the least

2
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java → functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/EmbeddedEngineExecutorService.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium.it.custom;
package org.springframework.cloud.fn.common.debezium;
import java.io.IOException;
import java.util.concurrent.ExecutorService;

1
functions/common/debezium-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports Normal file
View File

@@ -0,0 +1 @@
org.springframework.cloud.fn.common.debezium.DebeziumEngineBuilderAutoConfiguration

43
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java → functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium.it.custom;
package org.springframework.cloud.fn.common.debezium;
import java.io.File;
import java.time.Duration;
@@ -24,7 +24,11 @@ import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import javax.sql.DataSource;
import com.zaxxer.hikari.HikariDataSource;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.DebeziumEngine.Builder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -38,11 +42,12 @@ import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.fn.supplier.debezium.it.TestJdbcTemplateConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.jdbc.JdbcTestUtils;
@@ -50,7 +55,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
/**
* This test illustrate how to leverage the DebeziumEngineAutoConfiguration to build a supplier function with a custom
* This test illustrate how to leverage the DebeziumEngineAutoConfiguration to build a consumer function with a custom
* change event Consumer.
*
* @author Christian Tzolov
@@ -86,9 +91,9 @@ public class DebeziumEngineBuilderAutoConfigurationIntegrationTest {
"debezium.properties.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory", // new
"debezium.properties.schema.history.internal.file.filename=" + anotherTempDir.getAbsolutePath()
+ "schemahistory.dat", // new
+ "schemahistory.dat",
"debezium.properties.topic.prefix=my-topic", // new
"debezium.properties.topic.prefix=my-topic",
"debezium.properties.name=my-sql-connector",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
@@ -98,8 +103,6 @@ public class DebeziumEngineBuilderAutoConfigurationIntegrationTest {
"debezium.properties.database.hostname=localhost",
"debezium.properties.database.port=" + debeziumMySQL.getMappedPort(3306),
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector",
"debezium.properties.database.history=io.debezium.relational.history.MemoryDatabaseHistory",
// JdbcTemplate configuration
String.format("app.datasource.url=jdbc:mysql://localhost:%d/%s?enabledTLSProtocols=TLSv1.2",
@@ -140,12 +143,32 @@ public class DebeziumEngineBuilderAutoConfigurationIntegrationTest {
@SpringBootConfiguration
@EnableAutoConfiguration(exclude = { DataSourceAutoConfiguration.class, MongoAutoConfiguration.class })
@Import(TestJdbcTemplateConfiguration.class)
public static class DebeziumCustomConsumerApplication {
@Bean
public JdbcTemplate myJdbcTemplate(DataSource dataSource) {
return new JdbcTemplate(dataSource);
}
@Bean
@Primary
@ConfigurationProperties("app.datasource")
public DataSourceProperties dataSourceProperties() {
return new DataSourceProperties();
}
@Bean
public HikariDataSource dataSource(DataSourceProperties dataSourceProperties) {
return dataSourceProperties.initializeDataSourceBuilder()
.type(HikariDataSource.class)
.build();
}
@Bean
public EmbeddedEngineExecutorService embeddedEngine(Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer,
Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
DebeziumEngine<ChangeEvent<byte[], byte[]>> b = debeziumEngineBuilder.notifying(changeEventConsumer)
.build();
return new EmbeddedEngineExecutorService(debeziumEngineBuilder.notifying(changeEventConsumer).build());
}
@@ -171,7 +194,7 @@ public class DebeziumEngineBuilderAutoConfigurationIntegrationTest {
if (changeEvent != null) { // ignore null records
recordList.add(changeEvent);
keyValue.put(changeEvent.key(), changeEvent.value());
System.out.println("SIZE=" + recordList.size());
System.out.println("Key: " + recordList.size());
System.out.println("[Debezium Event]: " + changeEvent.toString());
}
}

2
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java → functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationTests.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium;
package org.springframework.cloud.fn.common.debezium;
import io.debezium.engine.DebeziumEngine;
import org.junit.jupiter.api.Test;

7
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java → functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumPropertiesTests.java
View File

@@ -14,12 +14,12 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium;
package org.springframework.cloud.fn.common.debezium;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumFormat;
import org.springframework.cloud.fn.supplier.debezium.DebeziumProperties.DebeziumOffsetCommitPolicy;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties.DebeziumFormat;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties.DebeziumOffsetCommitPolicy;
import static org.assertj.core.api.Assertions.assertThat;
@@ -36,7 +36,6 @@ public class DebeziumPropertiesTests {
public void defaultPropertiesTest() {
assertThat(this.properties.getPayloadFormat()).isEqualTo(DebeziumFormat.JSON);
assertThat(this.properties.getHeaderFormat()).isEqualTo(DebeziumFormat.JSON);
assertThat(this.properties.isCopyHeaders()).isEqualTo(true);
assertThat(this.properties.getOffsetCommitPolicy()).isEqualTo(DebeziumOffsetCommitPolicy.DEFAULT);
assertThat(this.properties.getProperties()).isNotNull();
assertThat(this.properties.getProperties()).isEmpty();

1
functions/common/pom.xml
View File

@@ -12,6 +12,7 @@
<modules>
<module>config-common</module>
<module>debezium-autoconfigure</module>
<module>aws-s3-common</module>
<module>file-common</module>
<module>ftp-common</module>

59
functions/supplier/debezium-supplier/pom.xml
View File

@@ -16,39 +16,13 @@
<properties>
<version.debezium>2.2.1.Final</version.debezium>
<apicurio.version>2.4.2.Final</apicurio.version>
</properties>
<dependencies>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-api</artifactId>
<version>${version.debezium}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-reload4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-embedded</artifactId>
<version>${version.debezium}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-reload4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>debezium-autoconfigure</artifactId>
<version>${project.version}</version>
</dependency>
<!-- Debezium Connectors -->
@@ -129,21 +103,6 @@
</exclusion>
</exclusions>
</dependency>
<!-- <dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-cassandra</artifactId>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>hibernate-validator</artifactId>
<groupId>org.hibernate</groupId>
</exclusion>
</exclusions>
<optional>true</optional>
</dependency> -->
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-spanner</artifactId>
@@ -157,18 +116,6 @@
<optional>true</optional>
</dependency>
<!-- APICURIO Schema Registry (for AVRO format)-->
<dependency>
<groupId>io.apicurio</groupId>
<artifactId>apicurio-registry-utils-converter</artifactId>
<version>${apicurio.version}</version>
</dependency>
<dependency>
<groupId>io.apicurio</groupId>
<artifactId>apicurio-registry-client</artifactId>
<version>${apicurio.version}</version>
</dependency>
<!-- TEST -->
<dependency>
<groupId>com.zaxxer</groupId>

16
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java
View File

@@ -34,10 +34,13 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.debezium.DebeziumEngineBuilderAutoConfiguration;
import org.springframework.cloud.fn.common.debezium.DebeziumProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
@@ -49,8 +52,9 @@ import org.springframework.util.MimeTypeUtils;
* @author Christian Tzolov
* @author Artem Bilan
*/
@Configuration
@EnableConfigurationProperties(DebeziumProperties.class)
@AutoConfiguration(after = DebeziumEngineBuilderAutoConfiguration.class)
@EnableConfigurationProperties({ DebeziumSupplierProperties.class })
@ConditionalOnBean(DebeziumEngine.Builder.class)
public class DebeziumReactiveConsumerConfiguration implements BeanClassLoaderAware {
private static final Log logger = LogFactory.getLog(DebeziumReactiveConsumerConfiguration.class);
@@ -103,9 +107,11 @@ public class DebeziumReactiveConsumerConfiguration implements BeanClassLoaderAwa
@Bean
@ConditionalOnMissingBean
public Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer(DebeziumProperties properties) {
public Consumer<ChangeEvent<byte[], byte[]>> changeEventConsumer(DebeziumProperties engineProperties,
DebeziumSupplierProperties supplierProperties) {
return new ChangeEventConsumer<byte[]>(properties.getPayloadFormat().contentType(), properties.isCopyHeaders(),
return new ChangeEventConsumer<byte[]>(engineProperties.getPayloadFormat().contentType(),
supplierProperties.isCopyHeaders(),
this.eventSink);
}

39
functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumSupplierProperties.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* Copyright 2023-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author Christian Tzolov
*/
@ConfigurationProperties("debezium.supplier")
public class DebeziumSupplierProperties {
/**
* Copy Change Event headers into Message headers.
*/
private boolean copyHeaders = true;
public boolean isCopyHeaders() {
return copyHeaders;
}
public void setCopyHeaders(boolean copyHeaders) {
this.copyHeaders = copyHeaders;
}
}

2
functions/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@@ -1 +1 @@
org.springframework.cloud.fn.supplier.debezium.DebeziumEngineBuilderAutoConfiguration
org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration

66
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumReactiveConsumerConfigurationTests.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* Copyright 2023-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium.it.supplier;
import io.debezium.engine.DebeziumEngine;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.fn.common.debezium.DebeziumEngineBuilderAutoConfiguration;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@link DebeziumReactiveConsumerConfiguration}.
*
* @author Christian Tzolov
*/
public class DebeziumReactiveConsumerConfigurationTests {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(DebeziumEngineBuilderAutoConfiguration.class,
DebeziumReactiveConsumerConfiguration.class));
// We have the debezium connectors on the classpath by default.
@Test
void noConnectorNoProperty() {
this.contextRunner.run((context) -> {
assertThat(context).doesNotHaveBean(DebeziumEngine.class);
});
}
@Test
void noConnectorWithProperty() {
this.contextRunner.withPropertyValues("debezium.properties.connector.class=Dummy")
.withClassLoader(new FilteredClassLoader("io.debezium.engine.DebeziumEngine$Builder"))
.run((context) -> {
assertThat(context).doesNotHaveBean(DebeziumEngine.class);
});
}
@Test
void withConnectorWithProperty() {
this.contextRunner.withPropertyValues("debezium.properties.connector.class=Dummy").run((context) -> {
assertThat(context).hasSingleBean(DebeziumEngine.class);
});
}
}

13
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java
View File

@@ -18,7 +18,6 @@ package org.springframework.cloud.fn.supplier.debezium.it.supplier;
import java.util.function.Supplier;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
@@ -27,12 +26,9 @@ import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.fn.supplier.debezium.it.TestJdbcTemplateConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
@@ -61,7 +57,6 @@ import static org.assertj.core.api.Assertions.assertThat;
"debezium.properties.topic.prefix=my-topic",
"debezium.properties.name=my-connector",
"debezium.properties.database.server.id=85744",
"debezium.properties.database.server.name=my-app-connector",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
"debezium.properties.database.user=debezium",
"debezium.properties.database.password=dbz",
@@ -96,8 +91,6 @@ public class DebeziumSupplierIntegrationTest {
debeziumMySQL.getMappedPort(3306), "inventory")); // JdbcTemplate config.
}
private ObjectMapper objectMapper = new ObjectMapper();
@Autowired
private Supplier<Flux<Message<?>>> debeziumSupplier;
@@ -109,7 +102,6 @@ public class DebeziumSupplierIntegrationTest {
jdbcTemplate.update(
"INSERT INTO `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
// jdbcTemplate.update("DELETE FROM `customers` WHERE `first_name` = ?", "Test666");
Flux<Message<?>> messageFlux = this.debeziumSupplier.get();
@@ -169,9 +161,8 @@ public class DebeziumSupplierIntegrationTest {
return new String((byte[]) message.getPayload());
}
@SpringBootApplication
@EnableAutoConfiguration(exclude = { MongoAutoConfiguration.class })
@Import({DebeziumReactiveConsumerConfiguration.class, TestJdbcTemplateConfiguration.class})
@SpringBootApplication(exclude = { MongoAutoConfiguration.class })
@Import({TestJdbcTemplateConfiguration.class})
static class DebeziumSupplierTestApplication {
}

2
functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java → functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/TestJdbcTemplateConfiguration.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.debezium.it;
package org.springframework.cloud.fn.supplier.debezium.it.supplier;
import javax.sql.DataSource;