Articles in this section

See more

Stream CDC Events into Message Queues

  • Updated
Follow

You can use Etlworks CDC Flows to stream real-time database changes—INSERTs, UPDATEs, and DELETEs—directly into message queues such as Kafka, Azure Event Hubs, Amazon Kinesis, and others. This is a scalable and high-performance pattern ideal for building event-driven architectures, pipelines, and microservices.

Captured CDC events are serialized as JSON (or Avro) and sent to the target topic or queue, where they can be consumed by downstream systems.

How It Works

Etlworks captures CDC events directly from the transaction log of the source database.

The CDC Flow sends each event to the destination message queue in real-time.

A separate consumer application can then: 

  • Read events from the queue,

  • Perform transformations, enrichments, or analytics,

  • Route the data to the appropriate destination. 

Note: The CDC Flow automatically performs an initial snapshot of all included tables before switching to real-time streaming.

When to Use This Flow Type

Use this Flow type when:

  • You want to stream CDC events into queues for real-time processing.

  • You already use Kafka, Event Hubs, or other messaging systems for data ingestion or pub/sub pipelines.

  • You want to decouple extraction from transformation and loading.

  • You are building microservices or event-driven data flows.

Prerequisites

Enable CDC in the source database:

Step-by-Step Guide

Step 1. Create a CDC Connection

Create a CDC connection for the source database. This connection will extract changes from the transaction log.

Step 2. Create a Message Queue Connection

Supported message queue destinations include:

  • Kafka (including Confluent, Aiven, CloudKafka, AWS MSK)

  • Azure Event Hubs

  • Amazon Kinesis

  • Amazon SQS

  • RabbitMQ

  • ActiveMQ

  • MQTT (IoT)

Refer to connector-specific articles to configure authentication, endpoints, and credentials.

Step 3. Create a Format

Create a JSON or Avro format. We recommend JSON unless you have specific downstream schema enforcement requirements.

Step 4. Create the CDC Flow

Go to Flows > Add Flow.

Choose Flow type:

➝ Stream CDC events into Message Queue

Step 5. Configure the Source (FROM)

In the newly created Flow:

  • Source Connection: use the CDC connection from Step 1.

  • FROM: choose which tables to include. You can use:

    • A comma-separated list of fully qualified table names (recommended).

    • An asterisk (*) includes all tables configured in the CDC connection.

    • A Regex expression to match table names.

    • A {token} → use tokenized values for dynamic configurations.

Example 1: Comma-separated tables

Select or enter tables or collections to stream data from.

FROM: inventory.city, inventory.country

Example 2: Asterisk (*)

Include all tables configured in the CDC connection:

FROM: *

Example 3: Regex match

Include only tables in the inventory schema whose names start with “c”:

FROM: ^inventory\.c[a-zA-Z0-9_]+

Example 4: {token}

Use a token for dynamic inclusion of tables (e.g., provided at runtime):

FROM: {included.tables}

When the Flow runs, {included.tables} is replaced with the actual list or pattern defined in your parameters/configuration.

Step 6. Configure the Destination (TO)

In the right-hand side (destination) of the transformation:

  • Destination Connection: use the queue connection from Step 2 (e.g., Kafka, Event Hub).

  • Format: use the JSON or Avro format from Step 3.

  • TO: override the default CDC Key to control how topics are named per table.

Step 7. Schedule the CDC Flow

We recommend using a continuous run schedule type.

  • The Flow runs indefinitely or until:

    • There are no more CDC events,

    • An error occurs,

  • After stopping, the Flow automatically restarts after a configured delay.

What is the CDC Key?

The CDC Key defines the topic or queue name to which CDC events are published.

By default, Etlworks uses one of the following CDC Key patterns:

  • [db]-[table]_cdc_stream — used when the schema is not relevant or available (e.g., in MySQL)

  • [db]_[schema]-[table]_cdc_stream — used when the schema is part of the fully qualified table name (e.g., in SQL Server)

Available Tokens

You can customize the CDC Key pattern using the following tokens:

  • [db] → Database name

  • [schema] → Schema name (if available)

  • [table] → Table name

  • [collection]→ Collection name (if available, e.q. in MongoDB)

Examples

Example 1: One topic per table

  • Database: test

  • Table: inventory

  • CDC Key: [table]

  • Topic: inventory

Example 2: One topic for all source tables

  • CDC Key: data

  • Topic: data

Use a Single Topic for All Tables

If you want all CDC events from all tables to be published to a single topic:

  1. Set the TO value to the topic name.

  2. Enable parameter Update fields for each event in the CDC connection to preserve context (table name, op, etc.).

  3. Choose the structure of the CDC event (see below).

Understanding the Message Structure

The message published to the queue directly depends on the Structure of the CDC Event setting defined in the CDC connection.

CDC events can include both data and metadata, such as:

  • before: row before the operation

  • after: row after the operation

  • source: metadata about the event

  • op: operation type (c = create, u = update, d = delete)

  • ts_ms: timestamp

  • transaction, schema, fields, etc.

IMPORTANT: If the Structure of the CDC Event option is left empty, Etlworks automatically flattens the event and includes only the after block in the JSON message sent to the queue. This configuration is recommended when the destination system expects only the current (post-operation) row state rather than the full CDC event envelope.

Generate Partition Keys for Kafka and Event Hub

To ensure order and consistency, we recommend setting a Partition Key in the CDC Connection.

Kafka and Event Hubs use the Partition Key to route messages to specific partitions.

If no key is set, events are randomly distributed, which may lead to out-of-order processing.

Examples:

  • [table] → events are partitioned per table

  • [table]-[op] → events are partitioned per table and operation

Supported Tokens for Partition Key

Each connector type supports a rich set of tokens. 

SQL Server tokens

  • [db]: database name.
  • [schema]: schema name.
  • [table]: table name.
  • [op]: the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms. 
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]: the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]: true for a snapshot.
  • [change_lsn]: change lns.
  • [commit_lsn]: commit lns.
  • [event_serial_no]: event serial number.

MySQL tokens

  • [db]: database name.
  • [table]: table name.
  • [op]:  the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms. 
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]: the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]: true for a snapshot.
  • [server_id]: the MySQL server ID.
  • [gtid]: the GTID.
  • [file]: the binlog filename.
  • [pos]: the position in the binlog.
  • [row]: the row number.

Oracle tokens

  • [source]: fully qualified event source.
  • [op]: the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms.
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]: the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]: true for a snapshot.
  • [txId]: the transaction id.
  • [scn]: the scn.

Postgres tokens

  • [db: database name.
  • [schema]: schema name.
  • [table]: table name.
  • [op]: the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms. 
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]:the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]:true for a snapshot.
  • [txId]: the transaction ID.
  • [lsn]: the lsn.
  • [xmin]: the xmin.

DB2 tokens

  • [db: database name.
  • [schema]: schema name.
  • [table]: table name.
  • [op]: the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms. 
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]:the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]:true for a snapshot.
  • [change_lsn]. 
  • [commit_lsn].

MongoDB tokens

  • [db]: the database name.
  • [collection]: the collection name.
  • [op]: the CDC operation: c (create), d (delete), u (update).
  • [timestamp]: the timestamp of the event in ms.
  • [version]: the connector version.
  • [connector]: the connector class name.
  • [name]: the connector name.
  • [ts_ms]: the timestamp of the event in ms.
  • [snapshot]: true for a snapshot.
  • [order]: the order of the event. 
  • [h]: the h of the event.

What’s Next

Once CDC events are published to the message queue, you can:

  • Consume them using microservices or stream processors (e.g., Kafka Streams, Flink).

  • Perform real-time transformations and enrichments.

  • Route events to warehouses, dashboards, or alerting systems.

Related Articles

Related articles