OpenLineage Integration

Debezium provides built-in integration with OpenLineage to automatically track data lineage for Change Data Capture (CDC) operations. The OpenLineage integration provides you with comprehensive visibility into the data flow and transformations that you use in your data pipeline.

About Data Lineage and OpenLineage

Data lineage tracks where data originates, how it moves between systems, and what transformations it undergoes. OpenLineage is the open standard that Debezium uses to emit and share that lineage metadata across your data pipeline.

Insights into data lineage is crucial for the following activities:

• Data governance and compliance
• Impact analysis when making changes
• Debugging data quality issues
• Understanding data dependencies

OpenLineage is an open standard for data lineage that provides a unified way to collect and track lineage metadata across multiple data systems. The specification defines a common model for describing datasets, jobs, and runs, simplifying the process of building comprehensive lineage graphs across heterogeneous data infrastructures.

For more information, see the OpenLineage website and documentation.

Run events

Debezium emits OpenLineage run events to track connector status throughout its lifecycle, from initialization through shutdown. These events provide visibility into connector health and enable real-time monitoring of data pipeline operations.

The connector emits run events after the following status changes:

START

Reports connector initialization.

RUNNING

Emitted periodically during normal streaming operations and during processing individual tables. These periodic events ensure continuous lineage tracking for long-running streaming CDC operations.

COMPLETE

Reports that the connector shut down gracefully.

FAIL

Reports that the connector encountered an error.

Deployment types

The Debezium OpenLineage integration is available for the following deployment types:

Kafka Connect

You run Debezium connectors as Kafka Connect plugins.

Debezium Server

You run Debezium in standalone server mode.

The same OpenLineage event model and features are available across the deployment types. But different processes are needed to configure the integration and to install dependencies.

How Debezium integrates with OpenLineage

Debezium maps its internal objects and events to the OpenLineage data model, representing connectors as Jobs and database tables or Kafka topics as Datasets. The following sections describe how those mappings work for source connectors, sink connectors, and the connector lifecycle.

OpenLineage job mapping

Each Debezium connector instance is represented as an OpenLineage Job, which captures the connector’s identity, version, and full configuration to support lineage tracing and pipeline debugging.

The Debezium connector is mapped to an OpenLineage Job, which includes the following elements:

Name

A job inherits its name from the Debezium topic.prefix, combined with the task ID (for example, inventory.0).

Namespace

Inherited from openlineage.integration.job.namespace, if specified; otherwise defaults to the value of the topic.prefix.

Debezium connector version

The version of the Debezium connector that is running and generating lineage events.

Complete connector configuration

All connector configuration properties, enabling full reproducibility and debugging of the data pipeline.

Job metadata

Description, tags, and owners.

Dataset mapping for source connectors

For source connectors, Debezium maps the database tables it monitors to OpenLineage input datasets and the Kafka topics it produces to output datasets, capturing schema information for both.

The following dataset mappings are possible:

Input Datasets

Represent the database tables that Debezium is configured to capture changes from. The OpenLineage integration automatically creates input datasets based on the connector configuration. The integration applies the following principles when it creates dataset mappings:

• Each table that the connector monitors becomes an input dataset.
• Each dataset captures schema information for the corresponding source table, including the name and data type of each column.
• DDL changes in the source table are reflected dynamically in the dataset schema.

Output datasets

Represent the Kafka topics where CDC events are written. For Kafka Connect deployments, output datasets are created when you apply the OpenLineage single message transformation (SMT). For Debezium Server deployments, output datasets are automatically captured.

Output dataset mappings are created according to the following principles:

• Each Kafka topic that the connector produces becomes an output dataset.
• The output dataset captures the complete CDC event structure, including metadata fields.
• The name of the dataset is based on the connector’s topic prefix configuration.

Dataset mapping for sink connectors

For sink connectors, the data flow is reversed compared to source connectors.

Input datasets

Represent the Kafka topics that the sink connector reads from. These topics typically contain CDC events from Debezium source connectors. The following principles apply when defining input datasets:

• Each Kafka topic that the sink connector consumes represents an input dataset.
• The input dataset specifies the Kafka topic schema and metadata.
• The namespace format follows kafka://bootstrap-server:port, where the bootstrap server is specified via the openlineage.integration.dataset.kafka.bootstrap.servers property.

Output datasets

Represent the target databases or collections where the sink connector writes data.

The following principles apply when defining output dataset mappings:

• Each target database table or collection represents an output dataset.
• The output dataset specifies the schema information for the target database.
• The namespace format depends on the target database system. For more information, see Dataset namespace formatting.

Sink connector availability in Kafka Connect vs. Debezium Server

The sink connectors that are available for OpenLineage integration vary with your deployment platform.

In a Kafka Connect environment, you can configure the following Debezium sink connectors to use the OpenLineage integration:

MongoDB sink connector

Writes CDC events to MongoDB collections.

JDBC sink connector

Writes CDC events to relational database tables.

In a Debezium Server environment, you can use the OpenLineage integration only with the Kafka sink. The MongoDB sink and JDBC sink connectors are not available for use with Debezium Server. Path: integrations/openlineage.adoc

Preparing to deploy the Debezium OpenLineage integration

Before you can use the Debezium OpenLineage integration, install the required dependencies for your deployment type — either Kafka Connect or Debezium Server.

Required Dependencies

The OpenLineage integration requires several JAR files that are bundled together in the debezium-openlineage-core-libs archive.

Kafka Connect

Before you can use Debezium with OpenLineage in Kafka Connect, you must obtain the required dependencies.

Procedure

1. Download the OpenLineage core archive.
2. Extract the contents of the archive into the Debezium plug-in directories in your Kafka Connect environment.

Debezium Server

Before you can use Debezium Server with OpenLineage, you must obtain the required dependencies.

Procedure

1. Download the OpenLineage core archive.
2. Extract the contents of the archive.
3. Copy all JAR files to the /debezium/lib directory in your Debezium Server installation.

Configuring the integration

To enable the integration, you must configure the Debezium connector and the OpenLineage client. The configuration approach differs between Kafka Connect and Debezium Server deployments.

Kafka Connect configuration

To enable Debezium to integrate with OpenLineage in Kafka Connect, add properties to your connector configuration, as shown in the following example:

# Enable OpenLineage integration
openlineage.integration.enabled=true

# Path to OpenLineage configuration file
openlineage.integration.config.file.path=/path/to/openlineage.yml

# Job metadata (optional but recommended)
openlineage.integration.job.namespace=myNamespace
openlineage.integration.job.description=CDC connector for products database
openlineage.integration.job.tags=env=prod,team=data-engineering
openlineage.integration.job.owners=Alice Smith=maintainer,Bob Johnson=Data Engineer

Debezium Server configuration

To enable Debezium Server to integrate with OpenLineage, add OpenLineage properties to the application.properties file, as shown in the following example. OpenLineage properties use the debezium.source. prefix.

# Enable OpenLineage integration
debezium.source.openlineage.integration.enabled=true

# Path to OpenLineage configuration file
debezium.source.openlineage.integration.config.file.path=config/openlineage.yml

# Job metadata (optional but recommended)
debezium.source.openlineage.integration.job.description=CDC connector for products database
debezium.source.openlineage.integration.job.tags=env=prod,team=data-engineering
debezium.source.openlineage.integration.job.owners=Alice Smith=maintainer,Bob Johnson=Data Engineer

Configuring the OpenLineage client

Create an openlineage.yml file to configure the OpenLineage client. The openlineage.yml configuration file is used in both Kafka Connect and Debezium Server deployments.

For Kafka Connect deployments, place the openlineage.yml file where the Kafka Connect worker nodes can read it. Similarly, for Debezium Server deployments, place the openlineage.yml file where the Debezium Server process can read it.

Common locations for storing the file on Kafka Connect include /kafka/openlineage.yml and /etc/debezium/openlineage.yml. For Debezium Server, it is typical to store the file in a subdirectory of the Debezium Server installation directory, for example, config/openlineage.yml. .Procedure . Create an openlineage.yml file in the appropriate location and configure it according to your needs using the following example as a guide:

+

transport:
  type: http
  url: http://your-openlineage-server:5000
  endpoint: /api/v1/lineage
  auth:
    type: api_key
    api_key: your-api-key

# Alternative: Console transport for testing
# transport:
#   type: console

1. Edit the config section of the deployment configuration YAML file for your platform to specify the openlineage.yml file path.
2. Restart the Debezium Server or Kafka Connect worker to apply the changes. For details about OpenLineage client configuration options, refer to the OpenLineage client documentation.

Debezium OpenLineage configuration properties

The following table lists the OpenLineage configuration properties for both deployment types.

Property (Kafka Connect) Description Required Default

openlineage.integration.enabled

Enables and disables the OpenLineage integration.

Yes

false

openlineage.integration.config.file.path

Path to the OpenLineage YAML configuration file.

Yes

No default value

openlineage.integration.job.namespace

Namespace used for the job.

No

Value of topic.prefix

openlineage.integration.job.description

Human-readable job description

No

No default value

openlineage.integration.job.tags

Comma-separated list of key-value tags.

No

No default value

openlineage.integration.job.owners

Comma-separated list of name-role ownership entries.

No

No default value

openlineage.integration.dataset.kafka.bootstrap.servers (source connectors only)

Kafka bootstrap servers used to retrieve Kafka topic metadata. For source connectors, if you do not specify a value, the value of schema.history.internal.kafka.bootstrap.servers is used.

For sink connectors, you must specify a value for this property.

Yes (for sink connectors)

Value of schema.history.internal.kafka.bootstrap.servers (for source connectors only)

Job metadata enrichment

To enhance the utility of the lineage data that it emits, you can add metadata that provides tags and owners information for each job. Tags are custom, arbitrary metadata that you can attach to jobs to provide additional context to help classify the job. Owners indicate who is responsible for the job. Use the following formats to specify tags and owners metadata in the OpenLineage configuration properties:

Tags list format

Specify tags as a comma-separated list of key-value pairs, as shown in the following example:

openlineage.integration.job.tags=environment=production,team=data-platform,criticality=high

Owners list format

Specify owners as a comma-separated list of name-role pairs, as shown in the following example:

openlineage.integration.job.owners=John Doe=maintainer,Jane Smith=Data Engineer,Team Lead=owner

Output dataset lineage

Debezium can capture output dataset lineage (Kafka topics) to track the destination of CDC events. The configuration approach differs between Kafka Connect and Debezium Server.

Kafka Connect output dataset lineage

To capture output dataset lineage in Kafka Connect, configure Debezium to use the OpenLineage Single Message Transform (SMT):

# Add OpenLineage transform
transforms=openlineage
transforms.openlineage.type=io.debezium.transforms.openlineage.OpenLineage

# Required: Configure schema history with Kafka bootstrap servers
schema.history.internal.kafka.bootstrap.servers=your-kafka:9092

The SMT captures detailed schema information about change events that Debezium writes to Kafka topics. The transformation captures schema data that includes the following items:

• Event structure (before, after, source, transaction metadata)
• Field types and nested structures
• Topic names and namespaces

Debezium Server output dataset lineage

For Debezium Server deployments, output dataset lineage is automatically captured when OpenLineage integration is enabled. No additional configuration or transformation is required, as Debezium Server has full control over the output records.

Complete configuration examples

The following examples show complete configurations for enabling OpenLineage integration in both Kafka Connect and Debezium Server.

Kafka Connect complete configuration example

The following example shows a complete configuration for enabling a PostgreSQL connector to integrate with OpenLineage in Kafka Connect:

{
  "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"
  }
}

Debezium Server complete configuration example

The following example shows a complete application.properties configuration for enabling a PostgreSQL connector to integrate with OpenLineage in Debezium Server with a Kafka sink:

# 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

MongoDB sink connector configuration example

The following example shows a complete configuration for enabling the MongoDB sink connector to integrate with OpenLineage in Kafka Connect:

{
  "name": "mongodb-sink",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbSinkConnector",
    "tasks.max": "1",
    "mongodb.connection.string": "mongodb://admin:admin@mongodb:27017",
    "topics": "inventory.inventory.products",
    "sink.database": "inventory2",
    "openlineage.integration.enabled": "true",
    "openlineage.integration.config.file.path": "/kafka/openlineage.yml",
    "openlineage.integration.job.description": "Sink connector for MongoDB",
    "openlineage.integration.job.tags": "env=prod,team=cdc",
    "openlineage.integration.job.owners": "Mario=maintainer,John Doe=Data scientist",
    "openlineage.integration.dataset.kafka.bootstrap.servers": "kafka:9092"
  }
}

JDBC sink connector configuration example

The following example shows a complete configuration for enabling the JDBC sink connector to integrate with OpenLineage in Kafka Connect:

{
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://postgres:5432/inventory",
    "connection.username": "postgres",
    "connection.password": "postgres",
    "topics": "inventory.inventory.customers",
    "insert.mode": "upsert",
    "primary.key.mode": "record_key",
    "openlineage.integration.enabled": "true",
    "openlineage.integration.config.file.path": "/kafka/openlineage.yml",
    "openlineage.integration.job.description": "Sink connector for JDBC",
    "openlineage.integration.job.tags": "env=prod,team=data-engineering",
    "openlineage.integration.job.owners": "Data Team=maintainer,Alice Johnson=Data Engineer",
    "openlineage.integration.dataset.kafka.bootstrap.servers": "kafka:9092"
  }
}

Dataset namespace formatting

Dataset namespaces identify the location and system of origin for input and output datasets. The namespace format varies by database system and connector type (source or sink), following the OpenLineage dataset naming specification.

Additional resources

• OpenLineage dataset naming specification

Input dataset namespaces

Input dataset namespaces identify the source database and follow a format specific to each database system. The following examples illustrate how Debezium formats input dataset namespaces for different database systems.

PostgreSQL input dataset (for source connectors)

• Namespace: postgres://hostname:port
• Name: schema.table
• Schema: Column names and types from the source table

Kafka input dataset (for sink connectors)

• Namespace: kafka://kafka-broker:9092
• Name: inventory.inventory.products
• Schema: CDC event structure from the source connector

The exact namespace format depends on your database system and follows the OpenLineage specification for dataset naming.

Output dataset namespaces for source connectors

Output dataset namespaces identify the Kafka topics where CDC events are written. The following example shows how Debezium formats the output dataset namespace for source connectors.

Kafka output dataset (for source connectors)

• Namespace: kafka://bootstrap-server:port
• Name: topic-prefix.schema.table
• Schema: Complete CDC event structure including metadata fields

Output dataset namespaces for sink connectors

Output dataset namespaces identify the target databases where sink connectors write data. The following examples show how Debezium sink connectors format output dataset namespaces for different database systems.

MongoDB output dataset

• Namespace: mongodb://mongodb-host:27017
• Name: database.collection
• Schema: Target collection schema

JDBC output dataset (PostgreSQL)

• Namespace: postgres://postgres-host:5432
• Name: schema.table
• Schema: Target table schema

Monitoring and Troubleshooting

You can verify that the Debezium OpenLineage integration is working correctly and diagnose problems by inspecting connector logs, reviewing OpenLineage events, and resolving common configuration issues.

Verifying the integration

To verify that the OpenLineage integration is working correctly, complete the following steps:

Procedure

1. Check the connector logs for OpenLineage-related messages.

2. If you configured HTTP transport, verify that events appear in your OpenLineage backend.

3. For testing purposes, you can configure console transport to view events directly in the logs, as shown in the following example:

   transport:
     type: console

Common issues

The following reference describes common problems that can prevent the OpenLineage integration from working correctly, along with steps to resolve them.

Integration not working

• Verify that openlineage.integration.enabled is set to true.
• Check that the path to the OpenLineage configuration file that is specified in the connector configuration is correct, and that Debezium can access the target file.
• Ensure that the YAML in the OpenLineage configuration file is valid.
• Verify that all required JAR dependencies are present in the classpath.

Missing output datasets

• Verify that you configured the connector to use the OpenLineage transformation.
• Check that you set the property schema.history.internal.kafka.bootstrap.servers in the connector configuration.

Connection issues

• Verify that you specified the correct server URL and authentication information in the OpenLineage client configuration.
• Check the network connectivity between Debezium and the OpenLineage server.

Dependency issues

• Ensure that all required JAR files are present and their versions are compatible versions.
• Check for classpath conflicts with existing dependencies.

Missing input datasets for sink connectors

• Verify that the openlineage.integration.dataset.kafka.bootstrap.servers property is configured.
• Verify that the connector has access to the Kafka bootstrap servers.
• Verify that the Kafka topics specified in the topics configuration exist and that the connector has access to them.

Error Events

When the connector fails, check for the following items in OpenLineage FAIL events:

• Error messages
• Stack traces
• Connector configuration for debugging