diff --git a/applications/source/debezium-source/README.adoc b/applications/source/debezium-source/README.adoc index ea4eeee0..94dac3bf 100644 --- a/applications/source/debezium-source/README.adoc +++ b/applications/source/debezium-source/README.adoc @@ -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$$:: $$$$ *($$Properties$$, default: `$$$$`)* -$$debezium.header-format$$:: $${@link ChangeEvent} header format. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)* -$$debezium.offset-commit-policy$$:: $$The policy that defines when the offsets should be committed to offset storage.$$ *($$DebeziumOffsetCommitPolicy$$, default: `$$$$`, possible values: `ALWAYS`,`PERIODIC`,`DEFAULT`)* -$$debezium.payload-format$$:: $${@link ChangeEvent} Key and Payload formats. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$$$`, 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$$, default: `$$$$`)* +Properties grouped by prefix: + + +=== debezium + +$$debezium-native-configuration$$:: $$$$ *($$Properties$$, default: `$$$$`)* +$$header-format$$:: $${@link ChangeEvent} header format. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$$$`, possible values: `JSON`,`AVRO`,`PROTOBUF`)* +$$offset-commit-policy$$:: $$The policy that defines when the offsets should be committed to offset storage.$$ *($$DebeziumOffsetCommitPolicy$$, default: `$$$$`, possible values: `ALWAYS`,`PERIODIC`,`DEFAULT`)* +$$payload-format$$:: $${@link ChangeEvent} Key and Payload formats. Defaults to 'JSON'.$$ *($$DebeziumFormat$$, default: `$$$$`, 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$$, default: `$$$$`)* + +=== 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 ---- diff --git a/applications/source/debezium-source/pom.xml b/applications/source/debezium-source/pom.xml index 14768a7e..083220b1 100644 --- a/applications/source/debezium-source/pom.xml +++ b/applications/source/debezium-source/pom.xml @@ -12,7 +12,6 @@ 1.25.1 - debezium-source diff --git a/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index 6fa81eab..6d4f1d1b 100644 --- a/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -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 diff --git a/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties index 6fa81eab..6d4f1d1b 100644 --- a/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ b/applications/source/debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -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 diff --git a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/databases/DebeziumDatabasesIntegrationTest.java b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/databases/DebeziumDatabasesIntegrationTest.java index 13e6efe4..1ae8a263 100644 --- a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/databases/DebeziumDatabasesIntegrationTest.java +++ b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/databases/DebeziumDatabasesIntegrationTest.java @@ -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[.]*")) { diff --git a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumDeleteHandlingIntegrationTest.java b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumDeleteHandlingIntegrationTest.java index fab10b61..6b331df8 100644 --- a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumDeleteHandlingIntegrationTest.java +++ b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumDeleteHandlingIntegrationTest.java @@ -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", diff --git a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumFlatteningIntegrationTest.java b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumFlatteningIntegrationTest.java index 60aa9fb7..9f95965a 100644 --- a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumFlatteningIntegrationTest.java +++ b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumFlatteningIntegrationTest.java @@ -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", diff --git a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumSupplierAvroFormatTest.java b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumSupplierAvroFormatTest.java index fa20e2fa..0bf1e6a7 100644 --- a/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumSupplierAvroFormatTest.java +++ b/applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumSupplierAvroFormatTest.java @@ -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", diff --git a/functions/common/debezium-autoconfigure/README.adoc b/functions/common/debezium-autoconfigure/README.adoc new file mode 100644 index 00000000..914f2831 --- /dev/null +++ b/functions/common/debezium-autoconfigure/README.adoc @@ -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 <> and <>, register a custom <> 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 +---- + + org.springframework.cloud.fn + debezium-autoconfigure + ${project-version} + +---- +[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 +---- + + io.debezium + debezium-connector-postgres + ${debezium-version} + + +---- +[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` (or `ChangeConsumer`) 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> 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> customConsumer() { // <1> + return new Consumer>() { + @Override + public void accept(ChangeEvent changeEvent) { + if (changeEvent != null) { // ignore null records + System.out.println("Key:" + changeEvent.key() + ", Value: " changeEvent.value()); + } + } + } +} + +@Bean +public DebeziumEngine> debeziumEngine( // <2> + Consumer> consumer, + DebeziumEngine.Builder> builder) { + + return new builder.notifying(consumer).build(); +} + +@Bean +public EmbeddedEngineExecutorService embeddedEngine( // <3> + DebeziumEngine> 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>>` 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$$, default: `$$$$`)*. +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` 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` 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> builder) { + + return new EmbeddedEngineExecutorService( // <3> + + builder.notifying(new ChangeConsumer<>() { // <1> + + @Override + public void handleBatch( + List> changeEventBatch, + RecordCommitter> committer) + throws InterruptedException { + + for (ChangeEvent 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>` provided by the auto-configuration. diff --git a/functions/common/debezium-autoconfigure/pom.xml b/functions/common/debezium-autoconfigure/pom.xml new file mode 100644 index 00000000..0202e65a --- /dev/null +++ b/functions/common/debezium-autoconfigure/pom.xml @@ -0,0 +1,88 @@ + + + 4.0.0 + + + org.springframework.cloud.fn + spring-functions-parent + 4.0.0-SNAPSHOT + ../../spring-functions-parent/pom.xml + + + debezium-autoconfigure + debezium-autoconfigure + Debezium Spring Boot auto-configuration + + + 2.2.1.Final + 2.4.2.Final + + + + + io.debezium + debezium-embedded + ${version.debezium} + + + slf4j-reload4j + org.slf4j + + + slf4j-log4j12 + org.slf4j + + + + + + + io.apicurio + apicurio-registry-utils-converter + ${apicurio.version} + + + io.apicurio + apicurio-registry-client + ${apicurio.version} + + + + + io.debezium + debezium-connector-mysql + ${version.debezium} + + + slf4j-log4j12 + org.slf4j + + + test + + + com.zaxxer + HikariCP + 4.0.3 + test + + + org.springframework + spring-jdbc + test + + + + + + org.apache.maven.plugins + maven-surefire-plugin + 3.0.0 + + integration + + + + + diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfiguration.java similarity index 97% rename from functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java rename to functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfiguration.java index dc42bba9..13243525 100644 --- a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfiguration.java +++ b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfiguration.java @@ -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> debeziumEngineBuilder( + public DebeziumEngine.Builder> debeziumEngineBuilder( OffsetCommitPolicy offsetCommitPolicy, CompletionCallback completionCallback, ConnectorCallback connectorCallback, DebeziumProperties properties, Clock debeziumClock) { diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumProperties.java similarity index 92% rename from functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java rename to functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumProperties.java index de625b88..23a2cf9b 100644 --- a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumProperties.java +++ b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/DebeziumProperties.java @@ -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 diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/EmbeddedEngineExecutorService.java similarity index 97% rename from functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java rename to functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/EmbeddedEngineExecutorService.java index c370943d..971e6cf0 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/EmbeddedEngineExecutorService.java +++ b/functions/common/debezium-autoconfigure/src/main/java/org/springframework/cloud/fn/common/debezium/EmbeddedEngineExecutorService.java @@ -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; diff --git a/functions/common/debezium-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/functions/common/debezium-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..a5f9a35b --- /dev/null +++ b/functions/common/debezium-autoconfigure/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +org.springframework.cloud.fn.common.debezium.DebeziumEngineBuilderAutoConfiguration diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java similarity index 85% rename from functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java rename to functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java index 2397eadc..b029ee6b 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/custom/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java +++ b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationIntegrationTest.java @@ -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> changeEventConsumer, Builder> debeziumEngineBuilder) { + DebeziumEngine> 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()); } } diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationTests.java similarity index 97% rename from functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java rename to functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationTests.java index 39f419a3..50840e8c 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumEngineBuilderAutoConfigurationTests.java +++ b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumEngineBuilderAutoConfigurationTests.java @@ -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; diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumPropertiesTests.java similarity index 87% rename from functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java rename to functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumPropertiesTests.java index b2314c24..6c963895 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/DebeziumPropertiesTests.java +++ b/functions/common/debezium-autoconfigure/src/test/java/org/springframework/cloud/fn/common/debezium/DebeziumPropertiesTests.java @@ -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(); diff --git a/functions/common/pom.xml b/functions/common/pom.xml index 9e3e6f39..95240601 100644 --- a/functions/common/pom.xml +++ b/functions/common/pom.xml @@ -12,6 +12,7 @@ config-common + debezium-autoconfigure aws-s3-common file-common ftp-common diff --git a/functions/supplier/debezium-supplier/pom.xml b/functions/supplier/debezium-supplier/pom.xml index 1a2d329a..b6089f0e 100644 --- a/functions/supplier/debezium-supplier/pom.xml +++ b/functions/supplier/debezium-supplier/pom.xml @@ -16,39 +16,13 @@ 2.2.1.Final - 2.4.2.Final - io.debezium - debezium-api - ${version.debezium} - - - slf4j-reload4j - org.slf4j - - - slf4j-log4j12 - org.slf4j - - - - - io.debezium - debezium-embedded - ${version.debezium} - - - slf4j-reload4j - org.slf4j - - - slf4j-log4j12 - org.slf4j - - + org.springframework.cloud.fn + debezium-autoconfigure + ${project.version} @@ -129,21 +103,6 @@ - io.debezium debezium-connector-spanner @@ -157,18 +116,6 @@ true - - - io.apicurio - apicurio-registry-utils-converter - ${apicurio.version} - - - io.apicurio - apicurio-registry-client - ${apicurio.version} - - com.zaxxer diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java index bdecf475..ed8ad039 100644 --- a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java +++ b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumReactiveConsumerConfiguration.java @@ -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> changeEventConsumer(DebeziumProperties properties) { + public Consumer> changeEventConsumer(DebeziumProperties engineProperties, + DebeziumSupplierProperties supplierProperties) { - return new ChangeEventConsumer(properties.getPayloadFormat().contentType(), properties.isCopyHeaders(), + return new ChangeEventConsumer(engineProperties.getPayloadFormat().contentType(), + supplierProperties.isCopyHeaders(), this.eventSink); } diff --git a/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumSupplierProperties.java b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumSupplierProperties.java new file mode 100644 index 00000000..55ab2018 --- /dev/null +++ b/functions/supplier/debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/debezium/DebeziumSupplierProperties.java @@ -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; + } +} diff --git a/functions/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/functions/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index e47f5f2a..2c4a031c 100644 --- a/functions/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/functions/supplier/debezium-supplier/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1 +1 @@ -org.springframework.cloud.fn.supplier.debezium.DebeziumEngineBuilderAutoConfiguration +org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumReactiveConsumerConfigurationTests.java b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumReactiveConsumerConfigurationTests.java new file mode 100644 index 00000000..a11cefcf --- /dev/null +++ b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumReactiveConsumerConfigurationTests.java @@ -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); + }); + } + +} diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java index 4bccf8fa..58655f23 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java +++ b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/DebeziumSupplierIntegrationTest.java @@ -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>> 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> 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 { } diff --git a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/TestJdbcTemplateConfiguration.java similarity index 96% rename from functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java rename to functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/TestJdbcTemplateConfiguration.java index 74a6841e..9e7918ae 100644 --- a/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/TestJdbcTemplateConfiguration.java +++ b/functions/supplier/debezium-supplier/src/test/java/org/springframework/cloud/fn/supplier/debezium/it/supplier/TestJdbcTemplateConfiguration.java @@ -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;