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

Refactor CDC Debezium Source

Browse Source 
- Add AVRO format support. Defaults to JSON
 - Remove the Metadata repo configuration.
 - Remove the Spring property sort-cuts. Only the native Debezium properties are used.
 - change the native properties prefix from `cdc.config to cdc.debezium`.
 - Removes the need to fork and modify the Debezium engine code.
 - Add binding name strategy.

Remove old CDC projects and replace with the new implementation
Remove duplicate pom properties
remove non existing pom module
Rename `cdc-debezium-supplier/source` to `debezium-supplier/source`
streamline naming and tests
streamline supplier tests
test streamline. Github workflow
Fix Debezium engine generics mess
Break `DebeziumConfiguration` into engine auto-config and consumer configuration
add initial reactive consumer impl.
Splits Debezium into a pure supplier and a streaming source

  - Remove any SCS related dependencies from the supplier.
  - Supplier uses Sinks.Many.

complete the supplier vs source separation
Improve Debezium Egine auto-conf

 - Provide configuration hooks for OffsetCommitPolicy, ConnectorCallback,
   CompletionCallback and Clock with default implementations.
 - Remove obsolete code.

minor improvements
Rename cdc property prefix to `debezium` and `cdc.debezium` to `debezium.inner`
remove obsolete `StreamBridge` tests
address reviews. improve testing
Supplier tests. Rename `inner` to `properties`.
Streamline the README. Minor improvements
streamline supplier tests
final fixes
This commit is contained in:
Christian Tzolov
2023-04-17 06:52:38 -04:00
committed by abilan
parent a363005677
commit 6cf66ec779
78 changed files with 2480 additions and 3854 deletions
Download Patch File Download Diff File Expand all files Collapse all files

339
applications/source/cdc-debezium-source/README.adoc
View File

@@ -1,339 +0,0 @@
//tag::ref-doc[]
= CDC Source
https://en.wikipedia.org/wiki/Change_data_capture[Change Data Capture] (CDC) `source` that captures and streams change events from various databases.
Currently, it supports `MySQL`, `PostgreSQL`, `MongoDB`, `Oracle` and `SQL Server` databases.
Build upon https://debezium.io/docs/embedded/[Debezium Embedded Connector], the `CDC Source` allows capturing and streaming database changes over different message binders such Apache Kafka, RabbitMQ and all Spring Cloud Stream supporter brokers.
It supports all Debezium configuration properties. Just add the `cdc.config.` prefix to the existing Debezium properties. For example to set the Debezium's `connector.class` property use the `cdc.config.connector.class` source property instead.
We provide convenient shortcuts for the most frequently used Debezium properties. For example instead of the long `cdc.config.connector.class=io.debezium.connector.mysql.MySqlConnector` Debezium property you can use our `cdc.connector=mysql` shortcut. The table below lists all available shortcuts along with the Debezium properties they represent.
The Debezium properties (e.g. `cdc.config.XXX`) always have precedence over the shortcuts!
The CDC Source introduces a new default `BackingOffsetStore` configuration, based on the https://github.com/spring-cloud/stream-applications/blob/main/functions/common/metadata-store-common/README.adoc[MetadataStore] service. Later provides various microservices friendly ways for storing the offset metadata.
== Options
//tag::configuration-properties[]
Properties grouped by prefix:
=== cdc
$$config$$:: $$Spring pass-trough wrapper for debezium configuration properties. All properties with a 'cdc.config.' prefix are native Debezium properties. The prefix is removed, converting them into Debezium io.debezium.config.Configuration.$$ *($$Map<String, String>$$, default: `$$<none>$$`)*
$$connector$$:: $$Shortcut for the cdc.config.connector.class property. Either of those can be used as long as they do not contradict with each other.$$ *($$ConnectorType$$, default: `$$<none>$$`, possible values: `mysql`,`postgres`,`mongodb`,`oracle`,`sqlserver`)*
$$name$$:: $$Unique name for this sourceConnector instance.$$ *($$String$$, default: `$$<none>$$`)*
$$schema$$:: $$Include the schema's as part of the outbound message.$$ *($$Boolean$$, default: `$$false$$`)*
=== cdc.flattening
$$add-fields$$:: $$Comma separated list of metadata fields to add to the flattened message. The fields will be prefixed with "__" or "__[<]struct]__", depending on the specification of the struct.$$ *($$String$$, default: `$$<none>$$`)*
$$add-headers$$:: $$Comma separated list specify a list of metadata fields to add to the header of the flattened message. The fields will be prefixed with "__" or "__[struct]__".$$ *($$String$$, default: `$$<none>$$`)*
$$delete-handling-mode$$:: $$Options for handling deleted records: (1) none - pass the records through, (2) drop - remove the records and (3) rewrite - add a '__deleted' field to the records.$$ *($$DeleteHandlingMode$$, default: `$$<none>$$`, possible values: `drop`,`rewrite`,`none`)*
$$drop-tombstones$$:: $$By default Debezium generates tombstone records to enable Kafka compaction on deleted records. The dropTombstones can suppress the tombstone records.$$ *($$Boolean$$, default: `$$true$$`)*
$$enabled$$:: $$Enable flattening the source record events (https://debezium.io/docs/configuration/event-flattening).$$ *($$Boolean$$, default: `$$true$$`)*
=== cdc.offset
$$commit-timeout$$:: $$Maximum number of milliseconds to wait for records to flush and partition offset data to be committed to offset storage before cancelling the process and restoring the offset data to be committed in a future attempt.$$ *($$Duration$$, default: `$$5000ms$$`)*
$$flush-interval$$:: $$Interval at which to try committing offsets. The default is 1 minute.$$ *($$Duration$$, default: `$$60000ms$$`)*
$$policy$$:: $$Offset storage commit policy.$$ *($$OffsetPolicy$$, default: `$$<none>$$`)*
$$storage$$:: $$Kafka connector tracks the number processed records and regularly stores the count (as "offsets") in a preconfigured metadata storage. On restart the connector resumes the reading from the last recorded source offset.$$ *($$OffsetStorageType$$, default: `$$<none>$$`, possible values: `memory`,`file`,`kafka`,`metadata`)*
=== cdc.stream.header
$$convert-connect-headers$$:: $$When true the {@link org.apache.kafka.connect.header.Header} are converted into message headers with the {@link org.apache.kafka.connect.header.Header#key()} as name and {@link org.apache.kafka.connect.header.Header#value()}.$$ *($$Boolean$$, default: `$$true$$`)*
$$offset$$:: $$Serializes the source record's offset metadata into the outbound message header under cdc.offset.$$ *($$Boolean$$, default: `$$false$$`)*
=== metadata.store.dynamo-db
$$create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)*
$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)*
$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
$$table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$<none>$$`)*
$$time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$<none>$$`)*
$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
=== metadata.store.jdbc
$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)*
$$table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$<none>$$`)*
=== metadata.store.mongo-db
$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)*
=== metadata.store.redis
$$key$$:: $$Redis key for metadata.$$ *($$String$$, default: `$$<none>$$`)*
=== metadata.store
$$type$$:: $$Indicates the type of metadata store to configure (default is 'memory'). You must include the corresponding Spring Integration dependency to use a persistent store.$$ *($$StoreType$$, default: `$$<none>$$`, possible values: `mongodb`,`redis`,`dynamodb`,`jdbc`,`zookeeper`,`hazelcast`,`memory`)*
=== metadata.store.zookeeper
$$connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)*
$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)*
$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)*
$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)*
//end::configuration-properties[]
==== Debezium property Shortcut mapping
The table below lists all available shortcuts along with the Debezium properties they represent.
.Table Shortcut Properties Mapping
|===
| Shortcut | Original | Description
|cdc.connector
|cdc.config.connector.class
|`mysql` : MySqlConnector, `postgres` : PostgresConnector, `mongodb` : MongodbSourceConnector, `oracle` : OracleConnector, `sqlserver` : SqlServerConnector
|cdc.name
|cdc.config.name
|
|cdc.offset.flush-interval
|cdc.config.offset.flush.interval.ms
|
|cdc.offset.commit-timeout
|cdc.config.offset.flush.timeout.ms
|
|cdc.offset.policy
|cdc.config.offset.commit.policy
|`periodic` : PeriodicCommitOffsetPolicy, `always` : AlwaysCommitOffsetPolicy
|cdc.offset.storage
|cdc.config.offset.storage
|`metadata` : MetadataStoreOffsetBackingStore, `file` : FileOffsetBackingStore, `kafka` : KafkaOffsetBackingStore, `memory` : MemoryOffsetBackingStore
|cdc.flattening.drop-tombstones
|cdc.config.drop.tombstones
|
|cdc.flattening.delete-handling-mode
|cdc.config.delete.handling.mode
|`none` : none, `drop` : drop, `rewrite` : rewrite
|===
== Database Support
The `CDC Source` uses the Debezium utilities, and currently supports CDC for five datastores: `MySQL`, `PostgreSQL`, `MongoDB`, `Oracle` and `SQL Server` databases.
== Examples and Testing
The [CdcSourceIntegrationTest](), [CdcDeleteHandlingIntegrationTest]() and [CdcFlatteningIntegrationTest]() integration tests use test databases fixtures, running on the local machine.
We use pre-build debezium docker database images.
The Maven builds create the test databases fixtures with the help of the `docker-maven-plugin`.
To run and debug the tests from your IDE you need to deploy the required database images from the command line.
Instructions below explains how to run pre-configured test databases form Docker images.
==== MySQL
Start the `debezium/example-mysql` in a docker:
[source, bash]
----
docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:1.0
----
[TIP]
====
(optional) Use `mysql` client to connected to the database and to create a `debezium` user with required credentials:
[source, bash]
----
docker run -it --rm --name mysqlterm --link mysql --rm mysql:5.7 sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -uroot -p"$MYSQL_ENV_MYSQL_ROOT_PASSWORD"'
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium' IDENTIFIED BY 'dbz';
----
====
Use following properties to connect the CDC Source to MySQL DB:
[source,properties]
----
cdc.connector=mysql # <1>
cdc.name=my-sql-connector # <2>
cdc.config.database.server.id=85744 # <2>
cdc.config.database.server.name=my-app-connector # <2>
cdc.config.database.user=debezium # <3>
cdc.config.database.password=dbz # <3>
cdc.config.database.hostname=localhost # <3>
cdc.config.database.port=3306 # <3>
cdc.schema=true # <4>
cdc.flattening.enabled=true # <5>
----
<1> Configures the CDC Source to use https://debezium.io/docs/connectors/mysql/[MySqlConnector]. (equivalent to setting `cdc.config.connector.class=io.debezium.connector.mysql.MySqlConnector`).
<2> Metadata used to identify and dispatch the incoming events.
<3> Connection to the MySQL server running on `localhost:3306` as `debezium` user.
<4> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `SourceRecord` events.
<5> Enables the https://debezium.io/docs/configuration/event-flattening/[CDC Event Flattening].
You can run also the `CdcSourceIntegrationTests#CdcMysqlTests` using this mysql configuration.
==== PostgreSQL
Start a pre-configured postgres server from the `debezium/example-postgres:1.0` Docker image:
[source, bash]
----
docker run -it --rm --name postgres -p 5432:5432 -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres debezium/example-postgres:1.0
----
You can connect to this server like this:
[source, bash]
----
psql -U postgres -h localhost -p 5432
----
Use following properties to connect the CDC Source to PostgreSQL:
[source,properties]
----
cdc.connector=postgres # <1>
cdc.offset.storage=memory #<2>
cdc.name=my-sql-connector # <3>
cdc.config.database.server.id=85744 # <3>
cdc.config.database.server.name=my-app-connector # <3>
cdc.config.database.user=postgres # <4>
cdc.config.database.password=postgres # <4>
cdc.config.database..dbname=postgres # <4>
cdc.config.database.hostname=localhost # <4>
cdc.config.database.port=5432 # <4>
cdc.schema=true # <5>
cdc.flattening.enabled=true # <6>
----
<1> Configures `CDC Source` to use https://debezium.io/docs/connectors/postgresql/[PostgresConnector]. Equivalent for setting `cdc.config.connector.class=io.debezium.connector.postgresql.PostgresConnector`.
<2> Configures the Debezium engine to use `memory` (e.g. `cdc.config.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore) backing offset store.
<3> Metadata used to identify and dispatch the incoming events.
<4> Connection to the PostgreSQL server running on `localhost:5432` as `postgres` user.
<5> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `SourceRecord` events.
<6> Enables the https://debezium.io/docs/configuration/event-flattening/[CDC Event Flattening].
You can run also the `CdcSourceIntegrationTests#CdcPostgresTests` using this mysql configuration.
==== MongoDB
Start a pre-configured mongodb from the `debezium/example-mongodb:0.10` Docker image:
[source, bash]
----
docker run -it --rm --name mongodb -p 27017:27017 -e MONGODB_USER=debezium -e MONGODB_PASSWORD=dbz debezium/example-mongodb:0.10
----
Initialize the inventory collections
[source, bash]
----
docker exec -it mongodb sh -c 'bash -c /usr/local/bin/init-inventory.sh'
----
In the `mongodb` terminal output, search for a log entry like `host: "3f95a8a6516e:27017"` :
[source, bash]
----
2019-01-10T13:46:10.004+0000 I COMMAND [conn1] command local.oplog.rs appName: "MongoDB Shell" command: replSetInitiate { replSetInitiate: { _id: "rs0", members: [ { _id: 0.0, host: "3f95a8a6516e:27017" } ] }, lsid: { id: UUID("5f477a16-d80d-41f2-9ab4-4ebecea46773") }, $db: "admin" } numYields:0 reslen:22 locks:{ Global: { acquireCount: { r: 36, w: 20, W: 2 }, acquireWaitCount: { W: 1 }, timeAcquiringMicros: { W: 312 } }, Database: { acquireCount: { r: 6, w: 4, W: 16 } }, Collection: { acquireCount: { r: 4, w: 2 } }, oplog: { acquireCount: { r: 2, w: 3 } } } protocol:op_msg 988ms
----
Add `127.0.0.1 3f95a8a6516e` entry to your `/etc/hosts`
Use following properties to connect the CDC Source to MongoDB:
[source,properties]
----
cdc.connector=mongodb # <1>
cdc.offset.storage=memory #<2>
cdc.config.mongodb.hosts=rs0/localhost:27017 # <3>
cdc.config.mongodb.name=dbserver1 # <3>
cdc.config.mongodb.user=debezium # <3>
cdc.config.mongodb.password=dbz # <3>
cdc.config.database.whitelist=inventory # <3>
cdc.config.tasks.max=1 # <4>
cdc.schema=true # <5>
cdc.flattening.enabled=true # <6>
----
<1> Configures `CDC Source` to use https://debezium.io/docs/connectors/mongodb/[MongoDB Connector]. This maps into `cdc.config.connector.class=io.debezium.connector.mongodb.MongodbSourceConnector`.
<2> Configures the Debezium engine to use `memory` (e.g. `cdc.config.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore) backing offset store.
<3> Connection to the MongoDB running on `localhost:27017` as `debezium` user.
<4> https://debezium.io/docs/connectors/mongodb/#tasks
<5> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `SourceRecord` events.
<6> Enables the https://debezium.io/docs/configuration/event-flattening/[CDC Event Flattening].
You can run also the `CdcSourceIntegrationTests#CdcPostgresTests` using this mysql configuration.
==== SQL Server
Start a `sqlserver` from the `debezium/example-postgres:1.0` Docker image:
[source, bash]
----
docker run -it --rm --name sqlserver -p 1433:1433 -e ACCEPT_EULA=Y -e MSSQL_PID=Standard -e SA_PASSWORD=Password! -e MSSQL_AGENT_ENABLED=true microsoft/mssql-server-linux:2017-CU9-GDR2
----
Populate with sample data form debezium's sqlserver tutorial:
[source, bash]
----
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-sqlserver-init/inventory.sql
cat ./inventory.sql | docker exec -i sqlserver bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD'
----
Use following properties to connect the CDC Source to SQLServer:
[source,properties]
----
cdc.connector=sqlserver # <1>
cdc.offset.storage=memory #<2>
cdc.name=my-sql-connector # <3>
cdc.config.database.server.id=85744 # <3>
cdc.config.database.server.name=my-app-connector # <3>
cdc.config.database.user=sa # <4>
cdc.config.database.password=Password! # <4>
cdc.config.database..dbname=testDB # <4>
cdc.config.database.hostname=localhost # <4>
cdc.config.database.port=1433 # <4>
----
<1> Configures `CDC Source` to use https://debezium.io/docs/connectors/sqlserver/[SqlServerConnector]. Equivalent for setting `cdc.config.connector.class=io.debezium.connector.sqlserver.SqlServerConnector`.
<2> Configures the Debezium engine to use `memory` (e.g. `cdc.config.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore) backing offset store.
<3> Metadata used to identify and dispatch the incoming events.
<4> Connection to the SQL Server running on `localhost:1433` as `sa` user.
You can run also the `CdcSourceIntegrationTests#CdcSqlServerTests` using this mysql configuration.
==== Oracle
Start Oracle reachable from localhost and set up with the configuration, users and grants described in the https://github.com/debezium/oracle-vagrant-box[Debezium Vagrant set-up]
Populate with sample data form Debezium's Oracle tutorial:
[source, bash]
----
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-with-oracle-jdbc/init/inventory.sql
cat ./inventory.sql | docker exec -i dbz_oracle sqlplus debezium/dbz@//localhost:1521/ORCLPDB1
----
//end::ref-doc[]
== Run standalone
[source,shell]
----
java -jar cdc-debezium-source.jar --cdc.connector=mysql --cdc.name=my-sql-connector --cdc.config.database.server.id=85744 --cdc.config.database.server.name=my-app-connector --cdc.config.database.user=debezium --cdc.config.database.password=dbz --cdc.config.database.hostname=localhost --cdc.config.database.port=3306 --cdc.schema=true --cdc.flattening.enabled=true
----

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

@@ -1,12 +0,0 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties, \
org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties$Header, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Flattening, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Offset, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \

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

@@ -1,12 +0,0 @@
configuration-properties.classes=org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties, \
org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties$Header, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Flattening, \
org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Offset, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \
org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \

142
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcDeleteHandlingIntegrationTest.java
View File

@@ -1,142 +0,0 @@
/*
* Copyright 2020-2020 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.stream.app.source.cdc;
import java.time.Duration;
import java.util.List;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
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.common.cdc.CdcCommonProperties;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.util.ClassUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL;
import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.CDC_SUPPLIER_OUT_0;
import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.receiveAll;
/**
* @author Christian Tzolov
* @author David Turanski
*/
@Testcontainers
public class CdcDeleteHandlingIntegrationTest extends CdcMySqlTestSupport {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.withPropertyValues(
"spring.cloud.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.schema=false",
"cdc.flattening.enabled=true",
"cdc.stream.header.offset=true",
"cdc.connector=mysql",
"cdc.config.database.user=debezium",
"cdc.config.database.password=dbz",
"cdc.config.database.hostname=localhost",
"cdc.config.database.port=" + MAPPED_PORT,
// "cdc.config.database.server.id=85744",
"cdc.config.database.server.name=my-app-connector",
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory");
@ParameterizedTest
@ValueSource(strings = {
"cdc.flattening.deleteHandlingMode=none,cdc.flattening.dropTombstones=true",
"cdc.flattening.deleteHandlingMode=none,cdc.flattening.dropTombstones=false",
"cdc.flattening.deleteHandlingMode=drop,cdc.flattening.dropTombstones=true",
"cdc.flattening.deleteHandlingMode=drop,cdc.flattening.dropTombstones=false",
"cdc.flattening.deleteHandlingMode=rewrite,cdc.flattening.dropTombstones=true",
"cdc.flattening.deleteHandlingMode=rewrite,cdc.flattening.dropTombstones=false"
})
public void handleRecordDeletions(String properties) {
contextRunner.withPropertyValues(properties.split(","))
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the
.run(consumer);
contextRunner.withPropertyValues(properties.split(","))
.run(consumer);
}
private String toString(Object object) {
return new String((byte[]) object);
}
final ContextConsumer<? super ApplicationContext> consumer = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
CdcCommonProperties props = context.getBean(CdcCommonProperties.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
CdcCommonProperties.DeleteHandlingMode deleteHandlingMode = props.getFlattening().getDeleteHandlingMode();
boolean isDropTombstones = props.getFlattening().isDropTombstones();
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
Message<?> received;
if (deleteHandlingMode == CdcCommonProperties.DeleteHandlingMode.drop) {
// Do nothing
}
else if (deleteHandlingMode == CdcCommonProperties.DeleteHandlingMode.none) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), CDC_SUPPLIER_OUT_0);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("null".getBytes());
}
else if (deleteHandlingMode == CdcCommonProperties.DeleteHandlingMode.rewrite) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), CDC_SUPPLIER_OUT_0);
assertThat(received).isNotNull();
assertThat(toString(received.getPayload()).contains("\"__deleted\":\"true\""));
}
if (!isDropTombstones && isKafkaPresent) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), CDC_SUPPLIER_OUT_0);
assertThat(received).isNotNull();
// Tombstones event should have KafkaNull payload
assertThat(received.getPayload().getClass().getCanonicalName())
.isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL);
String key = new String((byte[]) received.getHeaders().get("cdc_key"));
// Tombstones event should carry the deleted record id in the cdc_key header
assertThat(key).isEqualTo("{\"id\":" + newRecordId + "}");
}
received = outputDestination.receive(Duration.ofSeconds(1).toMillis(), CDC_SUPPLIER_OUT_0);
assertThat(received).isNull();
};
}

244
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcFlatteningIntegrationTest.java
View File

@@ -1,244 +0,0 @@
/*
* Copyright 2020-2021 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.stream.app.source.cdc;
import java.util.List;
import net.javacrumbs.jsonunit.core.Configuration;
import org.junit.jupiter.api.Test;
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.common.cdc.CdcCommonProperties;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import static net.javacrumbs.jsonunit.JsonAssert.assertJsonEquals;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL;
import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.receiveAll;
import static org.springframework.cloud.stream.app.source.cdc.CdcTestUtils.resourceToString;
/**
* @author Christian Tzolov
* @author David Turanski
* @author Artem Bilan
*/
public class CdcFlatteningIntegrationTest extends CdcMySqlTestSupport {
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.withPropertyValues(
"spring.cloud.function.definition=cdcSupplier",
"cdc.name=my-sql-connector",
"cdc.schema=false",
"cdc.stream.header.offset=false",
"cdc.connector=mysql",
"cdc.config.database.user=debezium",
"cdc.config.database.password=dbz",
"cdc.config.database.hostname=localhost",
"cdc.config.database.port=" + MAPPED_PORT,
// "cdc.config.database.server.id=85744",
"cdc.config.database.server.name=my-app-connector",
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory");
@Test
public void noFlattenedResponseNoKafka() {
contextRunner.withPropertyValues("cdc.flattening.enabled=false")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(noFlatteningTest);
}
@Test
public void noFlattenedResponseWithKafka() {
contextRunner.withPropertyValues("cdc.flattening.enabled=false")
.run(noFlatteningTest);
}
final ContextConsumer<? super ApplicationContext> noFlatteningTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
assertJsonEquals(resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()),
Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence", "source.ts_ms"));
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector");
assertJsonEquals("{\"databaseName\":\"inventory\"}", toString(messages.get(1).getHeaders().get("cdc_key")));
assertJsonEquals(resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()),
Configuration.empty().whenIgnoringPaths("source.sequence", "source.ts_ms"));
assertThat(messages.get(39).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.products");
assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key")));
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'");
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
messages = receiveAll(outputDestination);
assertThat(messages).hasSize(isKafkaPresent ? 4 : 3);
assertJsonEquals(resourceToString("classpath:/json/mysql_update_inventory_customers.json"),
toString(messages.get(1).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence"));
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
assertJsonEquals(resourceToString("classpath:/json/mysql_delete_inventory_customers.json"),
toString(messages.get(2).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence"));
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
if (isKafkaPresent) {
assertThat(messages.get(3).getPayload().getClass().getCanonicalName())
.isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
"Tombstones event should have KafkaNull payload");
assertThat(messages.get(3).getHeaders().get("cdc_topic"))
.isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key")));
}
};
@Test
public void flattenedResponseNoKafka() {
contextRunner.withPropertyValues(
"cdc.flattening.enabled=true",
"cdc.flattening.deleteHandlingMode=none",
"cdc.flattening.dropTombstones=false",
"cdc.flattening.addHeaders=op",
"cdc.flattening.addFields=name,db")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(flatteningTest);
}
@Test
public void flattenedResponseWithKafka() {
contextRunner.withPropertyValues(
"cdc.flattening.enabled=true",
"cdc.flattening.deleteHandlingMode=none",
"cdc.flattening.dropTombstones=false",
"cdc.flattening.addHeaders=op",
"cdc.flattening.addFields=name,db")
.run(flatteningTest);
}
@Test
public void flattenedResponseWithKafkaDropTombstone() {
contextRunner.withPropertyValues(
"cdc.flattening.enabled=true",
"cdc.flattening.deleteHandlingMode=none",
"cdc.flattening.dropTombstones=true",
"cdc.flattening.addHeaders=op",
"cdc.flattening.addFields=name,db")
.run(flatteningTest);
}
final ContextConsumer<? super ApplicationContext> flatteningTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
boolean isKafkaPresent = ClassUtils.isPresent(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
CdcCommonProperties.Flattening flatteningProps = context.getBean(CdcCommonProperties.class).getFlattening();
assertJsonEquals(resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()),
Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence", "source.ts_ms"));
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector");
assertJsonEquals("{\"databaseName\":\"inventory\"}",
toString(messages.get(1).getHeaders().get("cdc_key")));
if (flatteningProps.isEnabled()) {
assertJsonEquals(resourceToString("classpath:/json/mysql_flattened_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()));
}
else {
assertJsonEquals(resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()));
}
assertThat(messages.get(39).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.products");
assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("cdc_key")));
if (flatteningProps.isEnabled() && flatteningProps.getAddHeaders().contains("op")) {
assertThat(messages.get(39).getHeaders().get("__op")).isEqualTo("r");
}
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'");
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
messages = receiveAll(outputDestination);
assertThat(messages).hasSize((!flatteningProps.isDropTombstones() && isKafkaPresent) ? 4 : 3);
assertJsonEquals(resourceToString("classpath:/json/mysql_flattened_update_inventory_customers.json"),
toString(messages.get(1).getPayload()));
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
if (!StringUtils.isEmpty(flatteningProps.getAddHeaders()) && flatteningProps.getAddHeaders().contains("op")) {
assertThat(messages.get(1).getHeaders().get("__op")).isEqualTo("u");
}
if (flatteningProps.getDeleteHandlingMode() == CdcCommonProperties.DeleteHandlingMode.none) {
assertThat(toString(messages.get(2).getPayload())).isEqualTo("null");
assertThat(messages.get(1).getHeaders().get("cdc_topic")).isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(1).getHeaders().get("cdc_key")));
if (!StringUtils.isEmpty(flatteningProps.getAddHeaders())
&& flatteningProps.getAddHeaders().contains("op")) {
assertThat(messages.get(2).getHeaders().get("__op")).isEqualTo("d");
}
}
if (!flatteningProps.isDropTombstones() && isKafkaPresent) {
assertThat(messages.get(3).getPayload().getClass().getCanonicalName())
.isEqualTo(ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
"Tombstones event should have KafkaNull payload");
assertThat(messages.get(3).getHeaders().get("cdc_topic"))
.isEqualTo("my-app-connector.inventory.customers");
assertJsonEquals("{\"id\":" + newRecordId + "}", toString(messages.get(3).getHeaders().get("cdc_key")));
}
};
private String toString(Object object) {
return new String((byte[]) object);
}
}

62
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcMySqlTestSupport.java
View File

@@ -1,62 +0,0 @@
/*
* Copyright 2020-2021 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.stream.app.source.cdc;
import java.time.Duration;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.testcontainers.containers.GenericContainer;
import org.springframework.jdbc.core.JdbcTemplate;
/**
* @author David Turanski
*/
@Tag("integration")
public abstract class CdcMySqlTestSupport {
static final String DATABASE_NAME = "inventory";
static String MAPPED_PORT;
static GenericContainer debeziumMySQL = new GenericContainer<>("debezium/example-mysql:1.9.6.Final")
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
// .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("mysql")))
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
static {
debeziumMySQL.start();
}
static JdbcTemplate jdbcTemplate;
@BeforeAll
static void setup() {
MAPPED_PORT = String.valueOf(debeziumMySQL.getMappedPort(3306));
jdbcTemplate = CdcTestUtils.jdbcTemplate(
"com.mysql.cj.jdbc.Driver",
"jdbc:mysql://localhost:" + MAPPED_PORT + "/" + DATABASE_NAME + "?enabledTLSProtocols=TLSv1.2",
"root",
"debezium");
}
}

205
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcSourceDatabasesIntegrationTest.java
View File

@@ -1,205 +0,0 @@
/*
* Copyright 2020-2022 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.stream.app.source.cdc;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.images.builder.ImageFromDockerfile;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.util.CollectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Christian Tzolov
* @author David Turanski
* @author Artem Bilan
*/
@Tag("integration")
public class CdcSourceDatabasesIntegrationTest {
private static final String DEBEZIUM_EXAMPLE_MONGODB_1_9_6_FINAL = "debezium/example-mongodb:1.9.6.Final";
private static final String DEBEZIUM_EXAMPLE_POSTGRES_1_9_6_FINAL = "debezium/example-postgres:1.9.6.Final";
private static final String DEBEZIUM_EXAMPLE_MYSQL_1_9_6_FINAL = "debezium/example-mysql:1.9.6.Final";
private static final Log logger = LogFactory.getLog(CdcSourceDatabasesIntegrationTest.class);
private final SpringApplicationBuilder applicationBuilder = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(TestCdcSourceApplication.class))
.web(WebApplicationType.NONE)
.properties("spring.cloud.function.definition=cdcSupplier",
"cdc.name=my-connector",
"cdc.flattening.dropTombstones=false",
"cdc.schema=false",
"cdc.flattening.enabled=true",
"cdc.stream.header.offset=true",
"cdc.config.database.server.name=my-app-connector",
"cdc.config.database.history=io.debezium.relational.history.MemoryDatabaseHistory");
@Test
public void mysql() {
GenericContainer debeziumMySQL = new GenericContainer<>(DEBEZIUM_EXAMPLE_MYSQL_1_9_6_FINAL)
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
// .withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("mysql")))
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
debeziumMySQL.start();
String MAPPED_PORT = String.valueOf(debeziumMySQL.getMappedPort(3306));
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=mysql",
"--cdc.config.database.user=debezium",
"--cdc.config.database.password=dbz",
"--cdc.config.database.hostname=localhost",
"--cdc.config.database.port=" + MAPPED_PORT)) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = CdcTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
// Message size should correspond to the number of insert statements in the sample inventor DB configured in the debezium/example-mysql:1.9.6.Final:
// https://github.com/debezium/container-images/blob/main/examples/mysql/1.9/inventory.sql
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
}
}
@Test
@Disabled
public void sqlServer() {
GenericContainer sqlServer = new GenericContainer(new ImageFromDockerfile()
.withFileFromClasspath("Dockerfile", "sqlserver/Dockerfile")
.withFileFromClasspath("import-data.sh", "sqlserver/import-data.sh")
.withFileFromClasspath("inventory.sql", "sqlserver/inventory.sql")
.withFileFromClasspath("entrypoint.sh", "sqlserver/entrypoint.sh"))
.withEnv("ACCEPT_EULA", "Y")
.withEnv("MSSQL_PID", "Standard")
.withEnv("SA_PASSWORD", "Password!")
.withEnv("MSSQL_AGENT_ENABLED", "true")
.withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("sqlServer")))
.withExposedPorts(1433)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
//sqlServer.waitingFor(Wait.forLogMessage(".*(1 rows affected).*", 50)).start();
//sqlServer.waitingFor(Wait.forLogMessage(".*(Service Broker manager has started).*", 50)).start();
sqlServer.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=sqlserver",
// "--cdc.config.database.user=Standard",
"--cdc.config.database.user=sa",
"--cdc.config.database.password=Password!",
"--cdc.config.database.dbname=testDB",
"--cdc.config.database.hostname=localhost",
"--cdc.config.database.port=" + sqlServer.getMappedPort(1433))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = CdcTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
assertThat(messages).hasSize(30);
}
}
@Test
public void postgres() {
GenericContainer postgres = new GenericContainer(DEBEZIUM_EXAMPLE_POSTGRES_1_9_6_FINAL)
.withEnv("POSTGRES_USER", "postgres")
.withEnv("POSTGRES_PASSWORD", "postgres")
.withExposedPorts(5432)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
postgres.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=postgres",
"--cdc.config.database.user=postgres",
"--cdc.config.database.password=postgres",
"--cdc.config.slot.name=debezium",
"--cdc.config.database.dbname=postgres",
"--cdc.config.database.hostname=localhost",
// "--cdc.config.table.include.list=inventory.*",
"--cdc.config.database.port=" + postgres.getMappedPort(5432))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> allMessages = new ArrayList<>();
Awaitility.await().atMost(Duration.ofMinutes(5)).until(() -> {
List<Message<?>> messageChunk = CdcTestUtils.receiveAll(outputDestination);
if (!CollectionUtils.isEmpty(messageChunk)) {
logger.info("Chunk size: " + messageChunk.size());
allMessages.addAll(messageChunk);
}
// Message size should correspond to the number of insert statements in the sample inventor DB configured in the debezium/example-postgres:1.9.6.Final:
// https://github.com/debezium/container-images/blob/main/examples/postgres/1.9/inventory.sql
return allMessages.size() == 29; // Inventory DB entries
});
}
postgres.stop();
}
@Test
@Disabled
public void mongodb() {
GenericContainer mongodb = new GenericContainer(DEBEZIUM_EXAMPLE_MONGODB_1_9_6_FINAL)
.withEnv("MONGODB_USER", "debezium")
.withEnv("MONGODB_PASSWORD", "dbz")
.withExposedPorts(27017)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
mongodb.start();
try (ConfigurableApplicationContext context = applicationBuilder
.run("--cdc.connector=mongodb",
"--cdc.config.tasks.max=1",
"--cdc.config.mongodb.hosts=rs0/localhost:" + mongodb.getMappedPort(27017),
"--cdc.config.mongodb.name=dbserver1",
"--cdc.config.mongodb.user=debezium",
"--cdc.config.mongodb.password=dbz",
"--cdc.config.collection.include.list=inventory[.]*")) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = CdcTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
assertThat(messages).hasSize(666);
}
mongodb.stop();
}
}

32
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/TestCdcSourceApplication.java
View File

@@ -1,32 +0,0 @@
/*
* Copyright 2020-2020 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.stream.app.source.cdc;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration;
import org.springframework.context.annotation.Import;
/**
* @author Christian Tzolov
*/
@SpringBootConfiguration
@EnableAutoConfiguration(exclude = MongoAutoConfiguration.class)
@Import(CdcSupplierConfiguration.class)
public class TestCdcSourceApplication {
}

340
applications/source/debezium-source/README.adoc Normal file
View File

@@ -0,0 +1,340 @@
//tag::ref-doc[]
= Debezium Source
https://debezium.io/documentation/reference/2.2/development/engine.html[Debezium Engine] based https://en.wikipedia.org/wiki/Change_data_capture[Change Data Capture] (CDC) source.
The `Debezium Source` allows *capturing* database change events and *streaming* those over different message binders such `Apache Kafka`, `RabbitMQ` and all Spring Cloud Stream supporter brokers.
NOTE: This source can be used with *any* Spring Cloud Stream message binder.
It is not restricted nor depended on the Kafka Connect framework. Though this approach is flexible it comes with certain https://debezium.io/documentation/reference/2.2/development/engine.html#_handling_failures[limitations].
All Debezium configuration properties are supported.
Just precede any Debezium properties with the `debezium.properties.` prefix.
For example to set the Debezium's `connector.class` property use the `debezium.properties.connector.class` source property instead.
== Database Support
The `Debezium Source` currently supports CDC for multiple datastores: https://debezium.io/documentation/reference/2.2/connectors/mysql.html[MySQL], https://debezium.io/documentation/reference/2.2/connectors/postgresql.html[PostgreSQL], https://debezium.io/documentation/reference/2.2/connectors/mongodb.html[MongoDB], https://debezium.io/documentation/reference/2.2/connectors/oracle.html[Oracle], https://debezium.io/documentation/reference/2.2/connectors/sqlserver.html[SQL Server], https://debezium.io/documentation/reference/2.2/connectors/db2.html[Db2], https://debezium.io/documentation/reference/2.2/connectors/vitess.html[Vitess] and https://debezium.io/documentation/reference/2.2/connectors/spanner.html[Spanner] databases.
== 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.format$$:: $$Change Event message content 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.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>$$`)*
//end::configuration-properties[]
==== Event flattening configuration
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.
Using the https://debezium.io/documentation/reference/stable/transformations/event-flattening.html#_configuration[flattering configuration] you can configure simple messages format like this:
[source, bash]
----
--debezium.properties.transforms=unwrap
--debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
--debezium.properties.transforms.unwrap.drop.tombstones=false
--debezium.properties.transforms.unwrap.delete.handling.mode=rewrite
--debezium.properties.transforms.unwrap.add.fields=name,db
----
==== Connectors properties
The table below lists all available Debezium properties for each connecter.
Those properties can be used by prefixing them by the `debezium.properties.` prefix.
.Table of the configuration properties for every Debezium 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/cassandra.html[Cassandra]
// |https://debezium.io/documentation/reference/2.2/connectors/cassandra.html#cassandra-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
|===
== Examples and Testing
The debezium integration tests use databases fixtures, running on the local machine. Pre-build debezium docker database images with the help of Testcontainers are leveraged.
To run and debug the tests from your IDE you need to deploy the required database images from the command line.
Instructions below explains how to run pre-configured test databases form Docker images.
==== MySQL
Start the `debezium/example-mysql` in a docker:
[source, bash]
----
docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:2.2.0.Final
----
[TIP]
====
(optional) Use `mysql` client to connected to the database and to create a `debezium` user with required credentials:
[source, bash]
----
docker run -it --rm --name mysqlterm --link mysql --rm mysql:5.7 sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -uroot -p"$MYSQL_ENV_MYSQL_ROOT_PASSWORD"'
mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium' IDENTIFIED BY 'dbz';
----
====
Use following properties to connect the Debezium Source to MySQL DB:
[source,properties]
----
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.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>
debezium.properties.database.hostname=localhost # <3>
debezium.properties.database.port=3306 # <3>
debezium.properties.schema=true # <4>
debezium.properties.key.converter.schemas.enable=true # <4>
debezium.properties.value.converter.schemas.enable=true # <4>
debezium.properties.transforms=unwrap # <5>
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState # <5>
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>
----
<1> Configures the Debezium Source to use https://debezium.io/docs/connectors/mysql/[MySqlConnector].
<2> Metadata used to identify and dispatch the incoming events.
<3> Connection to the MySQL server running on `localhost:3306` as `debezium` user.
<4> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `ChangeEvent` message.
<5> Enables the https://debezium.io/documentation/reference/2.2/transformations/event-flattening.html[Change Event Flattening].
<6> Source state to preserver between multiple starts.
You can run also the `DebeziumDatabasesIntegrationTest#mysql()` using this mysql configuration.
NOTE: Disable the mysql GenericContainer test initialization code.
==== PostgreSQL
Start a pre-configured postgres server from the `debezium/example-postgres:1.0` Docker image:
[source, bash]
----
docker run -it --rm --name postgres -p 5432:5432 -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres debezium/example-postgres:2.2.0.Final
----
You can connect to this server like this:
[source, bash]
----
psql -U postgres -h localhost -p 5432
----
Use following properties to connect the Debezium Source to PostgreSQL:
[source,properties]
----
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>
debezium.properties.database..dbname=postgres # <4>
debezium.properties.database.hostname=localhost # <4>
debezium.properties.database.port=5432 # <4>
debezium.properties.schema=true # <5>
debezium.properties.key.converter.schemas.enable=true # <5>
debezium.properties.value.converter.schemas.enable=true # <5>
debezium.properties.transforms=unwrap # <6>
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState # <6>
debezium.properties.transforms.unwrap.add.fields=name,db # <6>
debezium.properties.transforms.unwrap.delete.handling.mode=none # <6>
debezium.properties.transforms.unwrap.drop.tombstones=true # <6>
----
<1> Configures `Debezium Source` to use https://debezium.io/docs/connectors/postgresql/[PostgresConnector].
<2> Configures the Debezium engine to use `memory` stores.
<3> Metadata used to identify and dispatch the incoming events.
<4> Connection to the PostgreSQL server running on `localhost:5432` as `postgres` user.
<5> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the message.
<6> Enables the https://debezium.io/docs/configuration/event-flattening/[Chage Event Flattening].
You can run also the `DebeziumDatabasesIntegrationTest#postgres()` using this postgres configuration.
NOTE: Disable the postgres GenericContainer test initialization code.
==== MongoDB
Start a pre-configured mongodb from the `debezium/example-mongodb:2.2.0.Final` container image:
[source, bash]
----
docker run -it --rm --name mongodb -p 27017:27017 -e MONGODB_USER=debezium -e MONGODB_PASSWORD=dbz debezium/example-mongodb:2.2.0.Final
----
Initialize the inventory collections
[source, bash]
----
docker exec -it mongodb sh -c 'bash -c /usr/local/bin/init-inventory.sh'
----
In the `mongodb` terminal output, search for a log entry like `host: "3f95a8a6516e:27017"` :
[source, bash]
----
2019-01-10T13:46:10.004+0000 I COMMAND [conn1] command local.oplog.rs appName: "MongoDB Shell" command: replSetInitiate { replSetInitiate: { _id: "rs0", members: [ { _id: 0.0, host: "3f95a8a6516e:27017" } ] }, lsid: { id: UUID("5f477a16-d80d-41f2-9ab4-4ebecea46773") }, $db: "admin" } numYields:0 reslen:22 locks:{ Global: { acquireCount: { r: 36, w: 20, W: 2 }, acquireWaitCount: { W: 1 }, timeAcquiringMicros: { W: 312 } }, Database: { acquireCount: { r: 6, w: 4, W: 16 } }, Collection: { acquireCount: { r: 4, w: 2 } }, oplog: { acquireCount: { r: 2, w: 3 } } } protocol:op_msg 988ms
----
Add `127.0.0.1 3f95a8a6516e` entry to your `/etc/hosts`
Use following properties to connect the Debezium Source to MongoDB:
[source,properties]
----
debezium.properties.connector.class=io.debezium.connector.mongodb.MongodbSourceConnector # <1>
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.mongodb.user=debezium # <3>
debezium.properties.mongodb.password=dbz # <3>
debezium.properties.database.whitelist=inventory # <3>
debezium.properties.tasks.max=1 # <4>
debezium.properties.schema=true # <5>
debezium.properties.key.converter.schemas.enable=true # <5>
debezium.properties.value.converter.schemas.enable=true # <5>
debezium.properties.transforms=unwrap # <6>
debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState # <6>
debezium.properties.transforms.unwrap.add.fields=name,db # <6>
debezium.properties.transforms.unwrap.delete.handling.mode=none # <6>
debezium.properties.transforms.unwrap.drop.tombstones=true # <6>
----
<1> Configures `Debezium Source` to use https://debezium.io/docs/connectors/mongodb/[MongoDB Connector].
<2> Configures the Debezium engine to use `memory`.
<3> Connection to the MongoDB running on `localhost:27017` as `debezium` user.
<4> https://debezium.io/docs/connectors/mongodb/#tasks
<5> Includes the https://debezium.io/docs/connectors/mysql/#change-events-value[Change Event Value] schema in the `SourceRecord` events.
<6> Enables the https://debezium.io/docs/configuration/event-flattening/[Chnage Event Flattening].
You can run also the `DebeziumDatabasesIntegrationTest#mongodb()` using this mongodb configuration.
==== SQL Server
Start a `sqlserver` from the `debezium/example-postgres:1.0` Docker image:
[source, bash]
----
docker run -it --rm --name sqlserver -p 1433:1433 -e ACCEPT_EULA=Y -e MSSQL_PID=Standard -e SA_PASSWORD=Password! -e MSSQL_AGENT_ENABLED=true microsoft/mssql-server-linux:2017-CU9-GDR2
----
Populate with sample data form debezium SqlServer tutorial:
[source, bash]
----
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-sqlserver-init/inventory.sql
cat ./inventory.sql | docker exec -i sqlserver bash -c '/opt/mssql-tools/bin/sqlcmd -U sa -P $SA_PASSWORD'
----
Use following properties to connect the Debezium Source to SQLServer:
[source,properties]
----
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>
debezium.properties.database..dbname=testDB # <4>
debezium.properties.database.hostname=localhost # <4>
debezium.properties.database.port=1433 # <4>
----
<1> Configures `Debezium Source` to use https://debezium.io/docs/connectors/sqlserver/[SqlServerConnector].
<2> Configures the Debezium engine to use `memory` state stores.
<3> Metadata used to identify and dispatch the incoming events.
<4> Connection to the SQL Server running on `localhost:1433` as `sa` user.
You can run also the `DebeziumDatabasesIntegrationTest#sqlServer()` using this SqlServer configuration.
==== Oracle
Start Oracle reachable from localhost and set up with the configuration, users and grants described in the https://github.com/debezium/oracle-vagrant-box[Debezium Vagrant set-up]
Populate with sample data form Debezium Oracle tutorial:
[source, bash]
----
wget https://raw.githubusercontent.com/debezium/debezium-examples/master/tutorial/debezium-with-oracle-jdbc/init/inventory.sql
cat ./inventory.sql | docker exec -i dbz_oracle sqlplus debezium/dbz@//localhost:1521/ORCLPDB1
----
//end::ref-doc[]
== Run standalone
[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
----

61
applications/source/cdc-debezium-source/pom.xml → applications/source/debezium-source/pom.xml
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0" 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>
@@ -12,20 +12,22 @@
<properties>
<json-unit.version>1.25.1</json-unit.version>
<mysql-connector-java.version>8.0.13</mysql-connector-java.version>
<!-- <mysql-connector-java.version>8.0.13</mysql-connector-java.version> -->
</properties>
<artifactId>cdc-debezium-source</artifactId>
<name>cdc-debezium-source</name>
<description>CDC Debezium source apps</description>
<artifactId>debezium-source</artifactId>
<name>debezium-source</name>
<description>Debezium source apps</description>
<packaging>jar</packaging>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>
<artifactId>debezium-supplier</artifactId>
<version>${java-functions.version}</version>
</dependency>
<!-- TEST -->
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
@@ -84,7 +86,40 @@
<artifactId>mysql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>4.0.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
<version>11.0.14</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>jakarta.servlet</groupId>
<artifactId>jakarta.servlet-api</artifactId>
<version>6.0.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-binder</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.33</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-testing-testcontainers</artifactId>
<version>2.2.0.Final</version>
</dependency>
</dependencies>
<build>
@@ -106,22 +141,22 @@
<artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
<configuration>
<application>
<name>cdc-debezium</name>
<name>debezium</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.cdc.CdcSupplierConfiguration.class
<configClass>org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration.class
</configClass>
<functionDefinition>cdcSupplier</functionDefinition>
<functionDefinition>debeziumSupplier</functionDefinition>
<properties>
<spring.autoconfigure.exclude>org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration</spring.autoconfigure.exclude>
<spring.cloud.stream.kafka.default.producer.messageKeyExpression>headers['cdc_key']</spring.cloud.stream.kafka.default.producer.messageKeyExpression>
<spring.cloud.stream.kafka.default.producer.messageKeyExpression>headers['debezium_key']</spring.cloud.stream.kafka.default.producer.messageKeyExpression>
</properties>
<maven>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>cdc-debezium-supplier</artifactId>
<artifactId>debezium-supplier</artifactId>
</dependency>
</dependencies>
</maven>

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

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

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

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

1
applications/source/debezium-source/src/main/resources/application.properties Normal file
View File

@@ -0,0 +1 @@
spring.cloud.stream.kafka.default.producer.messageKeyExpression=headers['debezium_key'].bytes

0
applications/source/cdc-debezium-source/src/test/docker/mongodb/Dockerfile → applications/source/debezium-source/src/test/docker/mongodb/Dockerfile
View File

0
applications/source/cdc-debezium-source/src/test/docker/mongodb/entrypoint.sh → applications/source/debezium-source/src/test/docker/mongodb/entrypoint.sh
View File

0
applications/source/cdc-debezium-source/src/test/docker/mongodb/import-data.sh → applications/source/debezium-source/src/test/docker/mongodb/import-data.sh
View File

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

@@ -0,0 +1,214 @@
/*
* Copyright 2020-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.stream.app.source.debezium.databases;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import io.debezium.testing.testcontainers.MongoDbContainer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.utility.DockerImageName;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.stream.app.source.debezium.integration.DebeziumTestUtils;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
import org.springframework.util.CollectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests integration with supported Debezium connector datastores. It uses the Debezium pre-build example-images for
* those datastores and pre-generated data for them.
*
* @author Christian Tzolov
* @author David Turanski
* @author Artem Bilan
*/
@Tag("integration")
public class DebeziumDatabasesIntegrationTest {
private static final Log logger = LogFactory.getLog(DebeziumDatabasesIntegrationTest.class);
private final SpringApplicationBuilder applicationBuilder = new SpringApplicationBuilder(
TestChannelBinderConfiguration
.getCompleteConfiguration(DebeziumDatabasesIntegrationTest.TestApplication.class))
.web(WebApplicationType.NONE)
.properties(
"spring.cloud.function.definition=debeziumSupplier",
// Flattening:
// https://debezium.io/documentation/reference/stable/transformations/event-flattening.html
"debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.drop.tombstones=false",
"debezium.properties.transforms.unwrap.delete.handling.mode=rewrite",
"debezium.properties.transforms.unwrap.add.fields=name,db,op,table",
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory",
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"debezium.properties.schema=false",
"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");
@Test
public void mysql() {
try (GenericContainer<?> mySQL = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MYSQL_IMAGE)
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3)) {
mySQL.start();
try (ConfigurableApplicationContext context = applicationBuilder.run(
"--debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
"--debezium.properties.database.user=debezium",
"--debezium.properties.database.password=dbz",
"--debezium.properties.database.hostname=localhost",
"--debezium.properties.database.port=" + mySQL.getMappedPort(3306))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
// Message size should correspond to the number of insert statements in:
// https://github.com/debezium/container-images/blob/main/examples/mysql/2.2/inventory.sql
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
}
mySQL.stop();
}
}
@Test
public void postgres() {
try (GenericContainer<?> postgres = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_POSTGRES_IMAGE)
.withEnv("POSTGRES_USER", "postgres")
.withEnv("POSTGRES_PASSWORD", "postgres")
.withExposedPorts(5432)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3)) {
postgres.start();
try (ConfigurableApplicationContext context = applicationBuilder.run(
"--debezium.properties.connector.class=io.debezium.connector.postgresql.PostgresConnector",
"--debezium.properties.database.user=postgres",
"--debezium.properties.database.password=postgres",
"--debezium.properties.slot.name=debezium",
"--debezium.properties.database.dbname=postgres",
"--debezium.properties.database.hostname=localhost",
"--debezium.properties.database.port=" + postgres.getMappedPort(5432))) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
List<Message<?>> allMessages = new ArrayList<>();
Awaitility.await().atMost(Duration.ofMinutes(5)).until(() -> {
List<Message<?>> messageChunk = DebeziumTestUtils.receiveAll(outputDestination);
if (!CollectionUtils.isEmpty(messageChunk)) {
logger.info("Chunk size: " + messageChunk.size());
allMessages.addAll(messageChunk);
}
// Message size should correspond to the number of insert statements in the sample inventor DB:
// https://github.com/debezium/container-images/blob/main/examples/postgres/2.2/inventory.sql
return allMessages.size() == 29; // Inventory DB entries
});
}
postgres.stop();
}
}
@Test
@Disabled
public void mongodb() {
GenericContainer<?> mongodb = MongoDbContainer
.node()
.imageName(DockerImageName.parse(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MONGODB_IMAGE))
.name("mymongo")
// .port(27017)
.replicaSet("rs0")
.skipDockerDesktopLogWarning(true)
.build();
// GenericContainer<?> mongodb = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MONGODB_IMAGE)
mongodb
.withEnv("MONGODB_USER", "debezium")
.withEnv("MONGODB_PASSWORD", "dbz")
.withExposedPorts(27017)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
mongodb.start();
String id = mongodb.getContainerId();
// String host = id.substring(0, 12);
String host = mongodb.getHost();
String port = "" + mongodb.getMappedPort(27017);
try (ConfigurableApplicationContext context = applicationBuilder
.run("--debezium.properties.connector.class=io.debezium.connector.mongodb.MongoDbConnector",
"--debezium.properties.topic.prefix=fullfillment",
"--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.mongodb.user=debezium",
"--debezium.properties.mongodb.password=dbz",
"--debezium.properties.collection.include.list=inventory[.]*")) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
// Number of entries should match the entries inserted by:
// https://github.com/debezium/container-images/blob/main/examples/mongodb/2.2/init-inventory.sh
assertThat(messages).hasSize(666);
}
mongodb.stop();
}
@SpringBootConfiguration
@EnableAutoConfiguration(exclude = { MongoAutoConfiguration.class, DataSourceAutoConfiguration.class })
@Import(DebeziumReactiveConsumerConfiguration.class)
public static class TestApplication {
}
}

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

@@ -0,0 +1,178 @@
/*
* Copyright 2020-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.stream.app.source.debezium.integration;
import java.time.Duration;
import java.util.List;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
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.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.util.ClassUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Christian Tzolov
* @author David Turanski
*/
@Tag("integration")
@Testcontainers
public class DebeziumDeleteHandlingIntegrationTest {
@Container
static GenericContainer<?> mySqlContainer = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MYSQL_IMAGE)
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
TestChannelBinderConfiguration.getCompleteConfiguration(TestDebeziumSourceApplication.class))
.withPropertyValues(
"spring.cloud.function.definition=debeziumSupplier",
"debezium.properties.schema=false",
"debezium.properties.key.converter.schemas.enable=false",
"debezium.properties.value.converter.schemas.enable=false",
"debezium.properties.topic.prefix=my-topic", // new
// 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.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"debezium.properties.name=my-connector",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
"debezium.properties.database.user=debezium",
"debezium.properties.database.password=dbz",
"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",
mySqlContainer.getMappedPort(3306), DebeziumTestUtils.DATABASE_NAME),
"app.datasource.username=root",
"app.datasource.password=debezium",
"app.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
"app.datasource.type=com.zaxxer.hikari.HikariDataSource");
@ParameterizedTest
@ValueSource(strings = {
"debezium.properties.transforms.unwrap.delete.handling.mode=none,debezium.properties.transforms.unwrap.drop.tombstones=true",
"debezium.properties.transforms.unwrap.delete.handling.mode=none,debezium.properties.transforms.unwrap.drop.tombstones=false",
"debezium.properties.transforms.unwrap.delete.handling.mode=drop,debezium.properties.transforms.unwrap.drop.tombstones=true",
"debezium.properties.transforms.unwrap.delete.handling.mode=drop,debezium.properties.transforms.unwrap.drop.tombstones=false",
"debezium.properties.transforms.unwrap.delete.handling.mode=rewrite,debezium.properties.transforms.unwrap.drop.tombstones=true",
"debezium.properties.transforms.unwrap.delete.handling.mode=rewrite,debezium.properties.transforms.unwrap.drop.tombstones=false"
})
public void handleRecordDeletions(String properties) {
contextRunner.withPropertyValues(properties.split(","))
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the
.run(consumer);
contextRunner.withPropertyValues(properties.split(","))
.run(consumer);
}
private String toString(Object object) {
return new String((byte[]) object);
}
final ContextConsumer<? super ApplicationContext> consumer = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class);
DebeziumProperties props = context.getBean(DebeziumProperties.class);
boolean isKafkaPresent = ClassUtils.isPresent(
DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
String deleteHandlingMode = props.getProperties().get("transforms.unwrap.delete.handling.mode");
String isDropTombstones = props.getProperties().get("transforms.unwrap.drop.tombstones");
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
Message<?> received;
if (deleteHandlingMode.equals("drop")) {
// Do nothing
}
else if (deleteHandlingMode.equals("none")) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), DebeziumTestUtils.BINDING_NAME);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("null".getBytes());
}
else if (deleteHandlingMode.equals("rewrite")) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), DebeziumTestUtils.BINDING_NAME);
assertThat(received).isNotNull();
assertThat(toString(received.getPayload()).contains("\"__deleted\":\"true\""));
}
if (!(isDropTombstones.equals("true")) && isKafkaPresent) {
received = outputDestination.receive(Duration.ofSeconds(10).toMillis(), DebeziumTestUtils.BINDING_NAME);
assertThat(received).isNotNull();
// Tombstones event should have KafkaNull payload
assertThat(received.getPayload().getClass().getCanonicalName())
.isEqualTo(DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL);
Object keyRaw = received.getHeaders().get("debezium_key");
String key = (keyRaw instanceof byte[]) ? new String((byte[]) keyRaw) : "" + keyRaw;
// Tombstones event should carry the deleted record id in the debezium_key header
assertThat(key).isEqualTo("{\"id\":" + newRecordId + "}");
}
received = outputDestination.receive(Duration.ofSeconds(1).toMillis(), DebeziumTestUtils.BINDING_NAME);
assertThat(received).isNull();
};
}

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

@@ -0,0 +1,301 @@
/*
* Copyright 2020-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.stream.app.source.debezium.integration;
import java.time.Duration;
import java.util.List;
import net.javacrumbs.jsonunit.JsonAssert;
import net.javacrumbs.jsonunit.core.Configuration;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
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.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.messaging.Message;
import org.springframework.test.jdbc.JdbcTestUtils;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Christian Tzolov
* @author David Turanski
* @author Artem Bilan
*/
@Testcontainers
@Tag("integration")
public class DebeziumFlatteningIntegrationTest {
@Container
static GenericContainer<?> mySqlContainer = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MYSQL_IMAGE)
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withUserConfiguration(
TestChannelBinderConfiguration.getCompleteConfiguration(TestDebeziumSourceApplication.class))
.withPropertyValues(
"spring.cloud.function.definition=debeziumSupplier",
"debezium.properties.schema=false",
"debezium.properties.key.converter.schemas.enable=false",
"debezium.properties.value.converter.schemas.enable=false",
"debezium.properties.topic.prefix=my-topic", // new
"debezium.properties.name=my-sql-connector",
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory", // new
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"debezium.properties.connector.class=io.debezium.connector.mysql.MySqlConnector",
"debezium.properties.database.user=debezium",
"debezium.properties.database.password=dbz",
"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",
mySqlContainer.getMappedPort(3306), DebeziumTestUtils.DATABASE_NAME),
"app.datasource.username=root",
"app.datasource.password=debezium",
"app.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
"app.datasource.type=com.zaxxer.hikari.HikariDataSource");
@Test
public void noFlattenedResponseNoKafka() {
contextRunner
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(noFlatteningTest);
}
@Test
public void noFlattenedResponseWithKafka() {
contextRunner.run(noFlatteningTest);
}
final ContextConsumer<? super ApplicationContext> noFlatteningTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class);
boolean isKafkaPresent = ClassUtils.isPresent(
DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
JsonAssert.assertJsonEquals(DebeziumTestUtils.resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()),
Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence",
"source.ts_ms", "ts_ms"));
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic");
JsonAssert.assertJsonEquals("{\"databaseName\":\"inventory\"}",
toString(messages.get(1).getHeaders().get("debezium_key")));
JsonAssert.assertJsonEquals(
DebeziumTestUtils.resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()),
Configuration.empty().whenIgnoringPaths("source.sequence", "source.ts_ms"));
assertThat(messages.get(39).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.products");
JsonAssert.assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("debezium_key")));
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'");
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).hasSize(isKafkaPresent ? 4 : 3);
JsonAssert.assertJsonEquals(
DebeziumTestUtils.resourceToString("classpath:/json/mysql_update_inventory_customers.json"),
toString(messages.get(1).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence"));
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(1).getHeaders().get("debezium_key")));
JsonAssert.assertJsonEquals(
DebeziumTestUtils.resourceToString("classpath:/json/mysql_delete_inventory_customers.json"),
toString(messages.get(2).getPayload()), Configuration.empty().whenIgnoringPaths("source.sequence"));
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(1).getHeaders().get("debezium_key")));
if (isKafkaPresent) {
assertThat(messages.get(3).getPayload().getClass().getCanonicalName())
.isEqualTo(DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
"Tombstones event should have KafkaNull payload");
assertThat(messages.get(3).getHeaders().get("debezium_destination"))
.isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(3).getHeaders().get("debezium_key")));
}
};
@Test
public void flattenedResponseNoKafka() {
contextRunner
.withPropertyValues("debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.add.fields=name,db,op",
"debezium.properties.transforms.unwrap.add.headers=name,op",
"debezium.properties.transforms.unwrap.delete.handling.mode=none",
"debezium.properties.transforms.unwrap.drop.tombstones=false")
.withClassLoader(new FilteredClassLoader(KafkaNull.class)) // Remove Kafka from the classpath
.run(flatteningTest);
}
@Test
public void flattenedResponseWithKafka() {
contextRunner
.withPropertyValues("debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.add.fields=name,db,op",
"debezium.properties.transforms.unwrap.add.headers=name,op",
"debezium.properties.transforms.unwrap.delete.handling.mode=none",
"debezium.properties.transforms.unwrap.drop.tombstones=false")
.run(flatteningTest);
}
@Test
public void flattenedResponseWithKafkaDropTombstone() {
contextRunner
.withPropertyValues("debezium.properties.transforms=unwrap",
"debezium.properties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState",
"debezium.properties.transforms.unwrap.add.fields=name,db,op",
"debezium.properties.transforms.unwrap.add.headers=name,op",
"debezium.properties.transforms.unwrap.delete.handling.mode=none",
"debezium.properties.transforms.unwrap.drop.tombstones=true")
.run(flatteningTest);
}
final ContextConsumer<? super ApplicationContext> flatteningTest = context -> {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class);
boolean isKafkaPresent = ClassUtils.isPresent(
DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
context.getClassLoader());
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
DebeziumProperties props = context.getBean(DebeziumProperties.class);
String deleteHandlingMode = props.getProperties().get("transforms.unwrap.delete.handling.mode");
String isDropTombstones = props.getProperties().get("transforms.unwrap.drop.tombstones");
JsonAssert.assertJsonEquals(DebeziumTestUtils.resourceToString(
"classpath:/json/mysql_ddl_drop_inventory_address_table.json"),
toString(messages.get(1).getPayload()),
Configuration.empty().whenIgnoringPaths("schemaName", "tableChanges", "source.sequence",
"source.ts_ms", "ts_ms"));
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic");
JsonAssert.assertJsonEquals("{\"databaseName\":\"inventory\"}",
toString(messages.get(1).getHeaders().get("debezium_key")));
if (isFlatteningEnabled(props)) {
JsonAssert.assertJsonEquals(
DebeziumTestUtils
.resourceToString("classpath:/json/mysql_flattened_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()));
}
else {
JsonAssert.assertJsonEquals(
DebeziumTestUtils.resourceToString("classpath:/json/mysql_insert_inventory_products_106.json"),
toString(messages.get(39).getPayload()));
}
assertThat(messages.get(39).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.products");
JsonAssert.assertJsonEquals("{\"id\":106}", toString(messages.get(39).getHeaders().get("debezium_key")));
jdbcTemplate.update(
"insert into `customers`(`first_name`,`last_name`,`email`) VALUES('Test666', 'Test666', 'Test666@spring.org')");
String newRecordId = jdbcTemplate.query("select * from `customers` where `first_name` = ?",
(rs, rowNum) -> rs.getString("id"), "Test666").iterator().next();
jdbcTemplate.update("UPDATE `customers` SET `last_name`='Test999' WHERE first_name = 'Test666'");
JdbcTestUtils.deleteFromTableWhere(jdbcTemplate, "customers", "first_name = ?", "Test666");
messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).hasSize((isDropTombstones.equals("false") && isKafkaPresent) ? 4 : 3);
JsonAssert.assertJsonEquals(
DebeziumTestUtils.resourceToString("classpath:/json/mysql_flattened_update_inventory_customers.json"),
toString(messages.get(1).getPayload()));
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(1).getHeaders().get("debezium_key")));
if (deleteHandlingMode.equals("none")) {
assertThat(toString(messages.get(2).getPayload())).isEqualTo("null");
assertThat(messages.get(1).getHeaders().get("debezium_destination")).isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(1).getHeaders().get("debezium_key")));
}
if (isDropTombstones.equals("false") && isKafkaPresent) {
assertThat(messages.get(3).getPayload().getClass().getCanonicalName())
.isEqualTo(DebeziumReactiveConsumerConfiguration.ORG_SPRINGFRAMEWORK_KAFKA_SUPPORT_KAFKA_NULL,
"Tombstones event should have KafkaNull payload");
assertThat(messages.get(3).getHeaders().get("debezium_destination"))
.isEqualTo("my-topic.inventory.customers");
JsonAssert.assertJsonEquals("{\"id\":" + newRecordId + "}",
toString(messages.get(3).getHeaders().get("debezium_key")));
}
};
private static boolean isFlatteningEnabled(DebeziumProperties props) {
String unwrapType = props.getProperties().get("transforms.unwrap.type");
return StringUtils.hasText(unwrapType) && unwrapType.equals("io.debezium.transforms.ExtractNewRecordState");
}
private String toString(Object object) {
if (object instanceof String) {
return (String) object;
}
return new String((byte[]) object);
}
}

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

@@ -0,0 +1,122 @@
/*
* 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.stream.app.source.debezium.integration;
import java.time.Duration;
import java.util.List;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.messaging.Message;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Christian Tzolov
*/
@Tag("integration")
@Testcontainers
public class DebeziumSupplierAvroFormatTest {
// E.g. docker run -it --rm --name apicurio -p 8080:8080 apicurio/apicurio-registry-mem:2.4.1.Final
@Container
static GenericContainer<?> apicurio = new GenericContainer<>("apicurio/apicurio-registry-mem:2.4.1.Final")
.withExposedPorts(8080)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
@Container
static GenericContainer<?> debeziumMySQL = new GenericContainer<>(DebeziumTestUtils.DEBEZIUM_EXAMPLE_MYSQL_IMAGE)
.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
.withEnv("MYSQL_USER", "mysqluser")
.withEnv("MYSQL_PASSWORD", "mysqlpw")
.withExposedPorts(3306)
.withStartupTimeout(Duration.ofSeconds(120))
.withStartupAttempts(3);
private final SpringApplicationBuilder applicationBuilder = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(TestDebeziumSourceApplication.class))
.web(WebApplicationType.NONE)
.properties(
"spring.cloud.function.definition=debeziumSupplier",
"debezium.format=AVRO",
"debezium.properties.key.converter=io.apicurio.registry.utils.converter.AvroConverter",
"debezium.properties.key.converter.apicurio.registry.auto-register=true",
"debezium.properties.key.converter.apicurio.registry.find-latest=true",
"debezium.properties.value.converter=io.apicurio.registry.utils.converter.AvroConverter",
"debezium.properties.value.converter.apicurio.registry.auto-register=true",
"debezium.properties.value.converter.apicurio.registry.find-latest=true",
"debezium.properties.schema.name.adjustment.mode=avro",
"debezium.properties.schema.history.internal=io.debezium.relational.history.MemorySchemaHistory",
"debezium.properties.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore",
"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",
"debezium.properties.database.hostname=localhost",
// JdbcTemplate configuration
String.format("app.datasource.url=jdbc:mysql://localhost:%d/%s?enabledTLSProtocols=TLSv1.2",
debeziumMySQL.getMappedPort(3306), DebeziumTestUtils.DATABASE_NAME),
"app.datasource.username=root",
"app.datasource.password=debezium",
"app.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
"app.datasource.type=com.zaxxer.hikari.HikariDataSource");
@Test
public void mysqlWithAvroContentFormat() {
String MYSQL_MAPPED_PORT = String.valueOf(debeziumMySQL.getMappedPort(3306));
String APICURIO_URL = "http://localhost:" + String.valueOf(apicurio.getMappedPort(8080)) + "/apis/registry/v2";
try (ConfigurableApplicationContext context = applicationBuilder.run(
"--debezium.properties.key.converter.apicurio.registry.url=" + APICURIO_URL,
"--debezium.properties.value.converter.apicurio.registry.url=" + APICURIO_URL,
"--debezium.properties.database.port=" + MYSQL_MAPPED_PORT)) {
OutputDestination outputDestination = context.getBean(OutputDestination.class);
// Using local region here
List<Message<?>> messages = DebeziumTestUtils.receiveAll(outputDestination);
assertThat(messages).isNotNull();
// Message size should correspond to the number of insert statements in the sample inventor DB
// configured by:
// https://github.com/debezium/container-images/blob/main/examples/mysql/2.1/inventory.sql
assertThat(messages).hasSizeGreaterThanOrEqualTo(52);
// assertThat(messages).map(message ->
// message.getHeaders().get("contentType")).isEqualTo("application/avro"); // TEST utils bug.
}
}
}

34
applications/source/cdc-debezium-source/src/test/java/org/springframework/cloud/stream/app/source/cdc/CdcTestUtils.java → applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/DebeziumTestUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.source.cdc;
package org.springframework.cloud.stream.app.source.debezium.integration;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@@ -24,38 +24,29 @@ import java.util.List;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.messaging.Message;
import org.springframework.util.StreamUtils;
/**
* @author Christian Tzolov
*/
public final class CdcTestUtils {
public final class DebeziumTestUtils {
/**
* derived from the spring.cloud.function.definition value
*/
public static final String CDC_SUPPLIER_OUT_0 = "cdcSupplier-out-0";
public static final String DATABASE_NAME = "inventory";
public static final String BINDING_NAME = "debeziumSupplier-out-0";
public static final String IMAGE_TAG = "2.2.0.Final";
public static final String DEBEZIUM_EXAMPLE_MYSQL_IMAGE = "debezium/example-mysql:" + IMAGE_TAG;
public static final String DEBEZIUM_EXAMPLE_POSTGRES_IMAGE = "debezium/example-postgres:" + IMAGE_TAG;
public static final String DEBEZIUM_EXAMPLE_MONGODB_IMAGE = "debezium/example-mongodb:" + IMAGE_TAG;
private CdcTestUtils() {
}
public static JdbcTemplate jdbcTemplate(String jdbcDriver, String jdbcUrl, String user, String password) {
private DebeziumTestUtils() {
DriverManagerDataSource dataSource = new DriverManagerDataSource();
dataSource.setDriverClassName(jdbcDriver);
dataSource.setUrl(jdbcUrl);
dataSource.setUsername(user);
dataSource.setPassword(password);
return new JdbcTemplate(dataSource);
}
public static List<Message<?>> receiveAll(OutputDestination outputDestination) {
return receiveAll(outputDestination, CDC_SUPPLIER_OUT_0);
return receiveAll(outputDestination, BINDING_NAME);
}
public static List<Message<?>> receiveAll(OutputDestination outputDestination, String bindingName) {
@@ -66,7 +57,8 @@ public final class CdcTestUtils {
if (received != null) {
list.add(received);
}
} while (received != null);
}
while (received != null);
return list;
}

62
applications/source/debezium-source/src/test/java/org/springframework/cloud/stream/app/source/debezium/integration/TestDebeziumSourceApplication.java Normal file
View File

@@ -0,0 +1,62 @@
/*
* Copyright 2020-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.stream.app.source.debezium.integration;
import javax.sql.DataSource;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.fn.supplier.debezium.DebeziumReactiveConsumerConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.core.JdbcTemplate;
/**
* @author Christian Tzolov
*/
@SpringBootConfiguration
@EnableAutoConfiguration(exclude = {MongoAutoConfiguration.class})
@Import(DebeziumReactiveConsumerConfiguration.class)
public class TestDebeziumSourceApplication {
@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)
//.driverClassName("com.mysql.cj.jdbc.Driver")
.build();
}
}

4
applications/source/cdc-debezium-source/src/test/resources/json/mysql_ddl_drop_inventory_address_table.json → applications/source/debezium-source/src/test/resources/json/mysql_ddl_drop_inventory_address_table.json
View File

@@ -2,8 +2,8 @@
"source": {
"version": "${json-unit.ignore}",
"connector": "mysql",
"name": "my-app-connector",
"ts_ms": 0,
"name": "my-topic",
"ts_ms": 1681662168850,
"snapshot": "true",
"db": "inventory",
"table": "addresses",

2
applications/source/cdc-debezium-source/src/test/resources/json/mysql_delete_inventory_customers.json → applications/source/debezium-source/src/test/resources/json/mysql_delete_inventory_customers.json
View File

@@ -9,7 +9,7 @@
"source": {
"version": "${json-unit.ignore}",
"connector": "mysql",
"name": "my-app-connector",
"name": "my-topic",
"ts_ms": "${json-unit.ignore}",
"snapshot": "false",
"db": "inventory",

3
applications/source/cdc-debezium-source/src/test/resources/json/mysql_flattened_insert_inventory_products_106.json → applications/source/debezium-source/src/test/resources/json/mysql_flattened_insert_inventory_products_106.json
View File

@@ -1,6 +1,7 @@
{
"__db" : "inventory",
"__name" : "my-app-connector",
"__name" : "my-topic",
"__op" : "r",
"id": 106,
"name": "hammer",
"description": "16oz carpenter's hammer",

5
applications/source/cdc-debezium-source/src/test/resources/json/mysql_flattened_update_inventory_customers.json → applications/source/debezium-source/src/test/resources/json/mysql_flattened_update_inventory_customers.json
View File

@@ -3,6 +3,7 @@
"first_name": "Test666",
"last_name": "Test999",
"email": "Test666@spring.org",
"__name": "my-app-connector",
"__db": "inventory"
"__name": "my-topic",
"__db": "inventory",
"__op": "u"
}

4
applications/source/cdc-debezium-source/src/test/resources/json/mysql_insert_inventory_products_106.json → applications/source/debezium-source/src/test/resources/json/mysql_insert_inventory_products_106.json
View File

@@ -9,8 +9,8 @@
"source": {
"version": "${json-unit.ignore}",
"connector": "mysql",
"name": "my-app-connector",
"ts_ms": 0,
"name": "my-topic",
"ts_ms": 1681663084000,
"snapshot": "true",
"db": "inventory",
"table": "products",

2
applications/source/cdc-debezium-source/src/test/resources/json/mysql_update_inventory_customers.json → applications/source/debezium-source/src/test/resources/json/mysql_update_inventory_customers.json
View File

@@ -14,7 +14,7 @@
"source": {
"version": "${json-unit.ignore}",
"connector": "mysql",
"name": "my-app-connector",
"name": "my-topic",
"ts_ms": "${json-unit.ignore}",
"snapshot": "false",
"db": "inventory",

0
applications/source/cdc-debezium-source/src/test/resources/logback-test.xml → applications/source/debezium-source/src/test/resources/logback-test.xml
View File

0
applications/source/cdc-debezium-source/src/test/resources/sqlserver/Dockerfile → applications/source/debezium-source/src/test/resources/sqlserver/Dockerfile
View File

0
applications/source/cdc-debezium-source/src/test/resources/sqlserver/entrypoint.sh → applications/source/debezium-source/src/test/resources/sqlserver/entrypoint.sh
View File

0
applications/source/cdc-debezium-source/src/test/resources/sqlserver/import-data.sh → applications/source/debezium-source/src/test/resources/sqlserver/import-data.sh
View File

0
applications/source/cdc-debezium-source/src/test/resources/sqlserver/inventory.sql → applications/source/debezium-source/src/test/resources/sqlserver/inventory.sql
View File

2
applications/source/pom.xml
View File

@@ -10,7 +10,7 @@
<packaging>pom</packaging>
<modules>
<module>cdc-debezium-source</module>
<module>debezium-source</module>
<module>file-source</module>
<module>ftp-source</module>
<module>jdbc-source</module>