### Debezium Server Startup Log Example Source: https://debezium.io/documentation/reference/3.5/operations/debezium-server.html This log output shows the Debezium Server starting up, including its ASCII art banner and informational messages about the Kinesis consumer and JSON converter configuration. ```log __ ____ __ _____ ___ __ ____ ______ --/ __ \/ / / / _ | / _ \/ //_/ / / / __/ -/ /_/ / /_/ / __ |/ , _/ ,< / /_/ /\ \ --\___\_\____/_/ |_/_/|_/_/|_|\____/___/ 2020-05-15 11:33:12,189 INFO [io.deb.ser.kin.KinesisChangeConsumer] (main) Using 'io.debezium.server.kinesis.KinesisChangeConsumer$$Lambda$119/0x0000000840130c40@f58853c' stream name mapper 2020-05-15 11:33:12,628 INFO [io.deb.ser.kin.KinesisChangeConsumer] (main) Using default KinesisClient 'software.amazon.awssdk.services.kinesis.DefaultKinesisClient@d1f74b8' 2020-05-15 11:33:12,628 INFO [io.deb.ser.DebeziumServer] (main) Consumer 'io.debezium.server.kinesis.KinesisChangeConsumer' instantiated 2020-05-15 11:33:12,754 INFO [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values: converter.type = key decimal.format = BASE64 schemas.cache.size = 1000 schemas.enable = true 2020-05-15 11:33:12,757 INFO [org.apa.kaf.con.jso.JsonConverterConfig] (main) JsonConverterConfig values: converter.type = value decimal.format = BASE64 schemas.cache.size = 1000 schemas.enable = false ``` -------------------------------- ### Debezium Connector Test Setup with Testcontainers Source: https://debezium.io/documentation/reference/3.5/integrations/testcontainers.html Set up Kafka, PostgreSQL, and Debezium containers for integration testing. Ensure all containers are started before proceeding. ```java public class DebeziumContainerTest { private static Network network = Network.newNetwork(); __**(1)** private static KafkaContainer kafkaContainer = new KafkaContainer() .withNetwork(network); __**(2)** public static PostgreSQLContainer postgresContainer = new PostgreSQLContainer<>( DockerImageName.parse("quay.io/debezium/postgres:15") .asCompatibleSubstituteFor("postgres")) .withNetwork(network) .withNetworkAliases("postgres"); __**(3)** public static DebeziumContainer debeziumContainer = new DebeziumContainer("quay.io/debezium/connect:3.5.2.Final") .withNetwork(network) .withKafka(kafkaContainer) .dependsOn(kafkaContainer); __**(4)** @BeforeClass public static void startContainers() { __**(5)** Startables.deepStart(Stream.of( kafkaContainer, postgresContainer, debeziumContainer)) .join(); } } ``` -------------------------------- ### Start minikube with insecure registry Source: https://debezium.io/documentation/reference/3.5/operations/kubernetes.html Use this command to start minikube with an insecure registry, which is necessary for pulling images within the cluster. ```bash $ minikube start --insecure-registry "10.0.0.0/24" ``` -------------------------------- ### Install Debezium Platform using OCI Artifact Source: https://debezium.io/documentation/reference/3.5/operations/debezium-platform.html Installs the Debezium platform using an OCI artifact. Similar to the Helm chart installation, it allows enabling the database for testing environments and requires the `domain.name` property for Ingress configuration. ```bash helm install debezium-platform --set database.enabled=true --set domain.name=platform.debezium.io oci://quay.io/debezium-charts/debezium-platform ``` -------------------------------- ### Full Ollama Provider Configuration Example Source: https://debezium.io/documentation/reference/3.5/ai/embeddings.html An example of a complete connector configuration when using the Ollama provider for embeddings. ```properties transforms=embeddings, transforms.embeddings.type=io.debezium.ai.embeddings.FieldToEmbedding transforms.embeddings.field.source=after.product transforms.embeddings.field.embedding=after.product_embedding transforms.embeddings.ollama.url=http://localhost:11434 transforms.embeddings.ollama.model.name=all-minilm ``` -------------------------------- ### Example: Enable ISBN Converter Source: https://debezium.io/documentation/reference/3.5/development/converters.html This example demonstrates how to enable a custom converter named 'isbn' with the fully qualified class name 'io.debezium.test.IsbnConverter'. ```properties converters: isbn isbn.type: io.debezium.test.IsbnConverter ``` -------------------------------- ### Example: Execute Incremental Snapshot Source: https://debezium.io/documentation/reference/3.5/connectors/sqlserver.html An example of inserting a signal into the Debezium signaling table to execute an incremental snapshot. This includes specific tables and an additional condition for filtering. ```sql INSERT INTO db1.myschema.debezium_signal (id, type, data) values ('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["db1.schema1.table1", "db1.schema1.table2"], "type":"incremental", "additional-conditions":[{"data-collection": "db1.schema1.table1" ,"filter":"color=\'blue\'"}]}'); ``` -------------------------------- ### Debezium Transaction Boundary Events Source: https://debezium.io/documentation/reference/3.5/connectors/vitess.html Example JSON structures for Debezium transaction boundary events, representing the start and end of a transaction. ```json { "status": "BEGIN", "id": "[\"{\\\"keyspace\\\":\\\"test_unsharded_keyspace\\\",\\\"shard\\\":\\\"0\\\",\\\"gtid\\\":\\\"MySQL56/e03ece6c-4c04-11ec-8e20-0242ac110004:1-37\\\"}]", "ts_ms": 1486500577000, "event_count": null, "data_collections": null } ``` ```json { "status": "END", "id": "[\"{\\\"keyspace\\\":\\\"test_unsharded_keyspace\\\",\\\"shard\\\":\\\"0\\\",\\\"gtid\\\":\\\"MySQL56/e03ece6c-4c04-11ec-8e20-0242ac110004:1-37\\\"}]", "ts_ms": 1486500577000, "event_count": 1, "data_collections": [ { "data_collection": "test_unsharded_keyspace.my_seq", "event_count": 1 } ] } ``` -------------------------------- ### Configure Custom Topic Creation Groups Source: https://debezium.io/documentation/reference/3.5/configuration/topic-auto-create-config.html This example shows how to define configurations for custom topic creation groups like 'inventory' and 'applicationlogs'. It includes include/exclude patterns, partition counts, cleanup policies, and retention periods. ```json { ... "topic.creation.inventory.include": "dbserver1\\.inventory\\.*", "topic.creation.inventory.partitions": 20, "topic.creation.inventory.cleanup.policy": "compact", "topic.creation.inventory.delete.retention.ms": 7776000000, "topic.creation.applicationlogs.include": "dbserver1\\.logs\\.applog-.*", "topic.creation.applicationlogs.exclude": "dbserver1\\.logs\\.applog-old-.*", "topic.creation.applicationlogs.replication.factor": 1, "topic.creation.applicationlogs.partitions": 20, "topic.creation.applicationlogs.cleanup.policy": "delete", "topic.creation.applicationlogs.retention.ms": 7776000000, "topic.creation.applicationlogs.compression.type": "lz4", ... ``` -------------------------------- ### JavaScript Filter Example Source: https://debezium.io/documentation/reference/3.5/transformations/filtering.html This JavaScript expression filters messages using the Struct#get() method. It selects update records with an 'id' of 2. ```JavaScript value.get('op') == 'u' && value.get('before').get('id') == 2 ``` -------------------------------- ### Complete Configuration for Default and Custom Topic Groups Source: https://debezium.io/documentation/reference/3.5/configuration/topic-auto-create-config.html Example of a complete connector configuration that includes settings for a default topic group and two custom topic creation groups: `inventory` and `applicationlogs`. This demonstrates how to define specific properties like replication factor, partitions, and cleanup policies for each group. ```json { ... "topic.creation.default.replication.factor": 3, "topic.creation.default.partitions": 10, "topic.creation.default.cleanup.policy": "compact", "topic.creation.default.compression.type": "lz4", "topic.creation.groups": "inventory,applicationlogs", "topic.creation.inventory.include": "dbserver1\.inventory\.*", ``` ```json "topic.creation.inventory.partitions": 20, "topic.creation.inventory.cleanup.policy": "compact", "topic.creation.inventory.delete.retention.ms": 7776000000, "topic.creation.applicationlogs.include": "dbserver1\.logs\.applog-.*", ``` ```json "topic.creation.applicationlogs.exclude": "dbserver1\.logs\.applog-old-.*", ``` ```json "topic.creation.applicationlogs.replication.factor": 1, "topic.creation.applicationlogs.partitions": 20, "topic.creation.applicationlogs.cleanup.policy": "delete", "topic.creation.applicationlogs.retention.ms": 7776000000, "topic.creation.applicationlogs.compression.type": "lz4" } ``` -------------------------------- ### Create TimescaleDB Hypertable Source: https://debezium.io/documentation/reference/3.5/transformations/timescaledb.html Example SQL command to create a 'conditions' hypertable in the 'public' schema. This demonstrates the setup required for TimescaleDB data capture. ```sql CREATE TABLE conditions (time TIMESTAMPTZ NOT NULL, location TEXT NOT NULL, temperature DOUBLE PRECISION NULL, humidity DOUBLE PRECISION NULL); SELECT create_hypertable('conditions', 'time'); ``` -------------------------------- ### Full Ehcache Configuration Example Source: https://debezium.io/documentation/reference/3.5/connectors/oracle.html An example demonstrating a complete Ehcache configuration, including global settings and specific configurations for various caches like transactions, processed transactions, schema changes, events, and rollbacks. ```json { "log.mining.buffer.type": "ehcache", "log.mining.buffer.ehcache.global.config": "", "log.mining.buffer.ehcache.transactions.config": "5121024000000", "log.mining.buffer.ehcache.processedtransactions.config": "5121024000000", "log.mining.buffer.ehcache.schemachanges.config": "5121024000000", "log.mining.buffer.ehcache.events.config": "5121024000000", "log.mining.buffer.ehcache.rollbacks.config": "5121024000000" } ``` -------------------------------- ### Kafka Connect Complete Configuration Example Source: https://debezium.io/documentation/reference/3.5/integrations/openlineage.html Example of a complete Kafka Connect configuration for a PostgreSQL connector with OpenLineage integration enabled. This includes connector settings and OpenLineage specific properties. ```json { "name": "inventory-connector-postgres", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "tasks.max": "1", "database.hostname": "postgres", "database.port": "5432", "database.user": "postgres", "database.password": "postgres", "database.dbname": "postgres", "topic.prefix": "inventory", "snapshot.mode": "initial", "slot.name": "inventory", "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", "schema.history.internal.kafka.topic": "schema-changes.inventory", "openlineage.integration.enabled": "true", "openlineage.integration.config.file.path": "/kafka/openlineage.yml", "openlineage.integration.job.description": "CDC connector for inventory database", "openlineage.integration.job.tags": "env=production,team=data-platform,database=postgresql", "openlineage.integration.job.owners": "Data Team=maintainer,Alice Johnson=Data Engineer", "transforms": "openlineage", "transforms.openlineage.type": "io.debezium.transforms.openlineage.OpenLineage" } } ``` -------------------------------- ### Enable JMX in Kafka Docker Container Source: https://debezium.io/documentation/reference/3.5/operations/monitoring.html Use this command to start a Kafka container with JMX enabled. Ensure JMXPORT and JMXHOST are set correctly for your environment. The example maps host port 9011 to the container's JMX port. ```bash $ docker run -it --rm --name kafka --hostname kafka \ -p 9092:9092 -p 9011:9011 \ -e CLUSTER_ID= \ -e NODE_ID=1 \ -e NODE_ROLE=combined \ -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \ -e KAFKA_LISTENERS=PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093 \ -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \ -e JMXPORT=9011 \ -e JMXHOST=10.0.1.10\ quay.io/debezium/kafka:{debezium-docker-label} ``` -------------------------------- ### Verify Kafka Startup Source: https://debezium.io/documentation/reference/3.5/tutorial.html Check this output to confirm that the Kafka server has started successfully and is ready for client connections. ```log ... 2025-07-22T11:43:00,935 - INFO [main:AppInfoParser$AppInfo@125] - Kafka version: 4.0.0 2025-07-22T11:43:00,938 - INFO [main:AppInfoParser$AppInfo@126] - Kafka commitId: 985bc99521dd22bb 2025-07-22T11:43:00,939 - INFO [main:AppInfoParser$AppInfo@127] - Kafka startTimeMs: 1753184580923 2025-07-22T11:43:00,959 - INFO [main:Logging@66] - [KafkaRaftServer nodeId=1] Kafka Server started ``` -------------------------------- ### Install Debezium SMT Go PDK Source: https://debezium.io/documentation/reference/3.5/transformations/filtering.html Command to obtain the Debezium SMT Go PDK. ```bash go get github.com/debezium/debezium-smt-go-pdk ``` -------------------------------- ### Debezium JDBC Connector Configuration Example Source: https://debezium.io/documentation/reference/3.5/connectors/jdbc.html This JSON configures the Debezium JDBC sink connector to consume from the 'orders' topic and write to a PostgreSQL database using upsert mode. Ensure the JDBC driver is correctly installed and accessible by Kafka Connect. ```json { "name": "jdbc-connector", "config": { "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector", "tasks.max": "1", "connection.url": "jdbc:postgresql://localhost/db", "connection.username": "pguser", "connection.password": "pgpassword", "insert.mode": "upsert", "delete.enabled": "true", "primary.key.mode": "record_key", "schema.evolution": "basic", "use.time.zone": "UTC", "topics": "orders" } } ``` -------------------------------- ### Install Debezium Management UDFs Source: https://debezium.io/documentation/reference/3.5/connectors/db2.html Connect to the database as the db2instl user and copy the Debezium management UDFs to the sqllib/function directory, then set permissions. ```bash db2 connect to DB_NAME ``` ```bash cp $HOME/asncdctools/src/asncdc $HOME/sqllib/function ``` ```bash chmod 777 $HOME/sqllib/function ``` -------------------------------- ### Debezium Change Event Field Schema with Propagation Source: https://debezium.io/documentation/reference/3.5/connectors/jdbc.html This example shows a field schema within a Debezium change event when column or data type propagation is enabled in the source connector. It includes parameters like '__debezium.source.column.type' and '__debezium.source.column.length' that guide the JDBC sink connector's mapping. ```json { "schema": { "type": "INT8", "parameters": { "__debezium.source.column.type": "TINYINT", "__debezium.source.column.length": "1" } } } ``` -------------------------------- ### Debezium Server Complete Configuration Example Source: https://debezium.io/documentation/reference/3.5/integrations/openlineage.html Example of a complete `application.properties` configuration for Debezium Server with a PostgreSQL connector and OpenLineage integration enabled for a Kafka sink. This includes sink, source, and OpenLineage specific settings. ```properties # Sink configuration (Kafka) debezium.sink.type=kafka debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer debezium.sink.kafka.producer.bootstrap.servers=kafka:9092 # Source connector configuration debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector debezium.source.offset.storage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore debezium.source.offset.flush.interval.ms=0 debezium.source.database.hostname=postgres debezium.source.database.port=5432 debezium.source.database.user=postgres debezium.source.database.password=postgres debezium.source.database.dbname=postgres debezium.source.topic.prefix=tutorial debezium.source.schema.include.list=inventory # OpenLineage integration debezium.source.openlineage.integration.enabled=true debezium.source.openlineage.integration.config.file.path=config/openlineage.yml debezium.source.openlineage.integration.job.description=CDC connector for products database debezium.source.openlineage.integration.job.tags=env=prod,team=cdc debezium.source.openlineage.integration.job.owners=Mario=maintainer,John Doe=Data scientist # Logging configuration (optional) quarkus.log.console.json=false ``` -------------------------------- ### Kafka Event Key Example Source: https://debezium.io/documentation/reference/3.5/connectors/mongodb-sink.html An example of an event key in a Kafka topic, which will be mapped to the MongoDB _id field. ```json { "userId": 1, "orderId": 1 } ``` -------------------------------- ### Verify MySQL Client Startup Source: https://debezium.io/documentation/reference/3.5/tutorial.html Confirms that the MySQL command line client has started. Note the warning about using passwords on the command line. ```log mysql: [Warning] Using a password on the command line interface can be insecure. Welcome to the MySQL monitor. Commands end with ; or \g. Your MySQL connection id is 9 Server version: 8.0.27 MySQL Community Server - GPL Copyright (c) 2000, 2021, Oracle and/or its affiliates. ``` -------------------------------- ### Verify MySQL Startup Source: https://debezium.io/documentation/reference/3.5/tutorial.html This output indicates that the MySQL server has started successfully and is ready to accept connections. ```log ... [System] [MY-010931] [Server] /usr/sbin/mysqld: ready for connections. Version: '8.0.27' socket: '/var/run/mysqld/mysqld.sock' port: 3306 MySQL Community Server - GPL. [System] [MY-011323] [Server] X Plugin ready for connections. Bind-address: '::' port: 33060, socket: /var/run/mysqld/mysqlx.sock ``` -------------------------------- ### Start Db2 Database Source: https://debezium.io/documentation/reference/3.5/connectors/db2.html Start the Db2 database if it is not already running. Replace DB_NAME with the actual database name. ```bash db2 start db DB_NAME ``` -------------------------------- ### Install Debezium Platform with Database Enabled Source: https://debezium.io/documentation/reference/3.5/operations/debezium-platform.html Installs the latest stable version of the Debezium platform using a Helm chart. The `--set database.enabled=true` flag automatically deploys a PostgreSQL database for the conductor service, simplifying testing environments. The `domain.name` property is required and sets the host for the Ingress definition. ```bash helm install debezium-platform debezium/debezium-platform --set database.enabled=true --set domain.name=platform.debezium.io ``` -------------------------------- ### Legacy ComputePartition Configuration Source: https://debezium.io/documentation/reference/3.5/transformations/partition-routing.html Example of the legacy ComputePartition SMT configuration for setting default partitions and replication factor. ```properties topic.creation.default.partitions=2 topic.creation.default.replication.factor=1 topic.prefix=fulfillment transforms=ComputePartition transforms.ComputePartition.type=io.debezium.transforms.partitions.ComputePartition transforms.ComputePartition.partition.data-collections.field.mappings=inventory.products:name,inventory.orders:purchaser transforms.ComputePartition.partition.data-collections.partition.num.mappings=inventory.products:2,inventory.orders:2 ``` -------------------------------- ### Deploy Apicurio Registry (In-Memory) Source: https://debezium.io/documentation/reference/3.5/configuration/avro.html Run an in-memory instance of Apicurio Registry. Ensure Docker is installed and you have the necessary permissions. ```bash docker run -it --rm --name apicurio \ -p 8080:8080 apicurio/apicurio-registry-mem:2.6.13.Final ``` -------------------------------- ### Debezium Initial Snapshot Notification (Started) Source: https://debezium.io/documentation/reference/3.5/configuration/notification.html Notification payload when an initial snapshot process has started. Includes connector name and timestamp. ```json { "id":"ff81ba59-15ea-42ae-b5d0-4d74f1f4038f", "aggregate_type":"Initial Snapshot", "type":"STARTED", "additional_data":{ "connector_name":"my-connector" }, "timestamp": "1695817046353" } ``` -------------------------------- ### Install PostgreSQL Decoderbufs Plugin on Fedora Source: https://debezium.io/documentation/reference/3.5/postgres-plugins.html Installs the Debezium PostgreSQL decoderbufs plugin using the DNF package manager on Fedora 30+. ```bash $ sudo dnf -y install postgres-decoderbufs ``` -------------------------------- ### Run MySQL Container with Docker Source: https://debezium.io/documentation/reference/3.5/tutorial.html Starts a MySQL container with the 'inventory' database preconfigured. It maps port 3306 and sets necessary user credentials. ```bash $ docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:3.5 ``` -------------------------------- ### Install Strimzi Operator via CLI Source: https://debezium.io/documentation/reference/3.5/operations/openshift.html Installs the Strimzi operator on OpenShift using a Subscription object. This operator manages Kafka deployments. ```yaml apiVersion: operators.coreos.com/v1alpha1 kind: Subscription metadata: name: my-strimzi-kafka-operator namespace: openshift-operators spec: channel: stable name: strimzi-kafka-operator source: operatorhubio-catalog sourceNamespace: olm ``` -------------------------------- ### Clone, Compile, and Install decoderbufs Source: https://debezium.io/documentation/reference/3.5/postgres-plugins.html This command sequence clones the decoderbufs repository, compiles the plugin, installs it, and then cleans up the source directory. Ensure you have the necessary build tools and PostgreSQL utilities like pg_config. ```bash $ git clone https://github.com/debezium/postgres-decoderbufs -b v{debezium-version} --single-branch \ && cd postgres-decoderbufs \ && make && make install \ && cd .. \ && rm -rf postgres-decoderbufs ``` -------------------------------- ### Example Outbox Message Source: https://debezium.io/documentation/reference/3.5/transformations/mongodb-outbox-event-router.html This is an example of a Kafka message generated by the MongoDB outbox event router SMT. It represents an outbox event for an order. ```json # Kafka Topic: outbox.event.order # Kafka Message key: "b2730779e1f596e275826f08" # Kafka Message Headers: "id=596e275826f08b2730779e1f" # Kafka Message Timestamp: 1556890294484 { "{\"id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}" } ``` -------------------------------- ### Run Kafka Connect Service with Docker Source: https://debezium.io/documentation/reference/3.5/tutorial.html Starts the Kafka Connect service in a container using the Debezium image. Ensure Kafka and MySQL are linked. Environment variables configure storage topics. ```bash $ docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link kafka:kafka --link mysql:mysql quay.io/debezium/connect:3.5 ``` -------------------------------- ### Schema Changes Signal Data Example Source: https://debezium.io/documentation/reference/3.5/connectors/oracle.html An example of the data structure for a 'schema-changes' signal, used to update table schemas in the Debezium Oracle connector. ```json { "database":"ORCLPDB1", "schema":"DEBEZIUM", "changes":[ { "type":"ALTER", "id":""ORCLPDB1"."DEBEZIUM"."CUSTOMER"", "table":{ "defaultCharsetName":null, "primaryKeyColumnNames":[ "ID", "NAME" ], "columns":[ { "name":"ID", "jdbcType":2, "typeName":"NUMBER", "typeExpression":"NUMBER", "charsetName":null, "length":9, "scale":0, "position":1, "optional":false, "autoIncremented":false, "generated":false }, { "name":"NAME", "jdbcType":12, "typeName":"VARCHAR2", "typeExpression":"VARCHAR2", "charsetName":null, "length":1000, "position":2 ```