Debezium Oracle CDC Source Connector

Type

source

Class

io.debezium.connector.oracle.OracleConnector

Target System

Oracle Database

Maintainer

Debezium / Red Hat

License

Apache License 2.0

Project

github.com/debezium/debezium

Download

Debezium Releases

This page documents version 3.0.8.Final. Newer versions should be compatible unless there are breaking changes, but field names or default values may differ. If you notice discrepancies, please contact Axual Support.

Description

The Debezium Oracle CDC Source Connector captures change events from Oracle databases using Oracle LogMiner and publishes them as records to Kafka topics.

It is maintained by the Debezium community as part of the open-source github.com/debezium/debezium.

The connector reads Oracle’s redo logs via the LogMiner API, which records every committed change. Schema history is stored in a dedicated Kafka topic so events are decoded correctly even after schema changes.

Features

Real-time Change Data Capture (CDC) from Oracle tables using LogMiner
Captures inserts, updates, and deletes
Schema history stored in a dedicated Kafka topic
Axual-specific SSL key material managed via keyvault references
Stream name follows the convention <nickname>.<schema>.<table>

When to Use

You need real-time CDC from Oracle tables into Kafka.
You want to capture all DML events (inserts, updates, deletes) with low latency.

When NOT to Use

You only need periodic polling — use the JDBC Source Connector — Oracle instead.
Oracle LogMiner is not available or supplemental logging is not enabled on your database.

Installation

The connector is available from the Debezium Releases.

The Debezium Oracle connector requires the Oracle JDBC driver to be placed in the connector’s plugin directory. The driver is not bundled due to licensing restrictions. Select the JAR matching your Java runtime version: ojdbc8.jar for Java 8, ojdbc11.jar for Java 11, or ojdbc17.jar for Java 17+. Obtain it from the Oracle JDBC Downloads page.

Navigate to the releases page and select the version matching your Kafka Connect installation.
Download the connector archive and extract the JAR files.
Download the Oracle JDBC driver (ojdbc8.jar for Java 8, ojdbc11.jar for Java 11, or ojdbc17.jar for Java 17+) from the Oracle JDBC Downloads page and add it to the connector plugin directory.

For installation steps, see Installing Connector Plugins.

Configuration

For the complete configuration reference, see the official source connector documentation.

To configure a connector in Axual Self-Service, see Starting Connectors.

Axual Platform requirements

The Debezium Oracle connector maintains a schema history by writing to and reading from a dedicated Kafka topic (schema.history.internal.kafka.topic). This means the connector opens an internal Kafka producer and consumer (running separately from the main Connect worker) to persist schema changes. On the Axual Platform, Kafka requires SSL authentication, so this internal connection must also be configured with the SSL credentials of the connector application. The SSL values are read from Axual’s keyvault at runtime using the ${keyvault:…} reference syntax.

Supply the following additional properties when running this connector on the Axual Platform.

The schema.history.internal.kafka.topic value must be resolved using Axual’s name-resolving mechanism. Consult an Axual platform operator to acquire the name-resolving values (tenant, instance, and environment short names).

auto.create.topics.enable

false

schema.history.internal.consumer.group.id

This consumer-group name must be resolved!

Example resolved value:
axual-local-default-oracle_nickname-dbhistory

schema.history.internal.consumer.security.protocol

SSL

schema.history.internal.consumer.ssl.key.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.key.password}

schema.history.internal.consumer.ssl.keystore.location

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.keystore.location}

schema.history.internal.consumer.ssl.keystore.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.keystore.password}

schema.history.internal.consumer.ssl.truststore.location

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.truststore.location}

schema.history.internal.consumer.ssl.truststore.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.truststore.password}

schema.history.internal.producer.security.protocol

SSL

schema.history.internal.producer.ssl.key.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.key.password}

schema.history.internal.producer.ssl.keystore.location

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.keystore.location}

schema.history.internal.producer.ssl.keystore.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.keystore.password}

schema.history.internal.producer.ssl.truststore.location

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.truststore.location}

schema.history.internal.producer.ssl.truststore.password

${keyvault:connectors/axual/local/default/my_oracle_app:ssl.truststore.password}

Getting Started

Prerequisites

Oracle Database instance

You need a running Oracle Database instance with LogMiner available and supplemental logging enabled.

Enable supplemental logging on the target database and tables:

-- Enable minimal supplemental logging at database level
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- Enable full supplemental logging for the target table
ALTER TABLE ORCLPDB1.HOTEL ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

Create a dedicated Debezium user with the required permissions:

CREATE USER c##dbzuser IDENTIFIED BY debezium
  DEFAULT TABLESPACE users
  QUOTA UNLIMITED ON users
  CONTAINER=ALL;

GRANT CREATE SESSION TO c##dbzuser CONTAINER=ALL;
GRANT SET CONTAINER TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$DATABASE TO c##dbzuser CONTAINER=ALL;
GRANT FLASHBACK ANY TABLE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ANY TABLE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL;
GRANT EXECUTE_CATALOG_ROLE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ANY TRANSACTION TO c##dbzuser CONTAINER=ALL;
GRANT LOGMINING TO c##dbzuser CONTAINER=ALL;
GRANT CREATE TABLE TO c##dbzuser CONTAINER=ALL;
GRANT LOCK ANY TABLE TO c##dbzuser CONTAINER=ALL;
GRANT CREATE SEQUENCE TO c##dbzuser CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR TO c##dbzuser CONTAINER=ALL;
GRANT EXECUTE ON DBMS_LOGMNR_D TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOG TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOG_HISTORY TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_LOGS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_CONTENTS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGMNR_PARAMETERS TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$LOGFILE TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVED_LOG TO c##dbzuser CONTAINER=ALL;
GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO c##dbzuser CONTAINER=ALL;

Axual name-resolving properties

Some properties require Axual’s custom name-resolving mechanism.
Consult an Axual platform operator to acquire the following information:

Tenant shortName

Example value:
axual

Instance shortName

Example value:
local

Environment shortName

Example value:
default

Connect-Application shortName

This value is already provided. We’ll use:
my_oracle_app

Resolving pattern for topic-names

Example pattern:
${tenant-shortName}-${instance-shortName}-${environment-shortName}-${topic-name}
Example resolved topic name:
axual-local-default-transactions

Resolving pattern for consumer-groups

Example pattern:
${tenant-shortName}-${instance-shortName}-${environment-shortName}-${consumerGroup-name}
Example resolved consumer group:
axual-local-default-com.axual.transactionProcessor

Axual streams

Follow the Creating streams documentation to create and deploy the following streams:

oracle_nickname._schema-changes — Stores the database schema history.
oracle_nickname.ORCLPDB1.HOTEL — Contains DB events for the HOTEL table.
Stream name follows the pattern <nickname>.<schema>.<table>. Create one per watched table.

All streams should use Key/Value type String/String, Partition count 1, Retention Time 157680000000 (5 years).

Steps

Step 1 — Create a connector application

Follow the Configure and install a connector documentation to set up a new Connector-Application.
Let’s call it my_oracle_app.
The plugin name is io.debezium.connector.oracle.OracleConnector.
If a plugin isn’t available, ask a platform operator to install plugins.

Use the following application ID: oracle_nickname-dbhistory.

Step 2 — Configure the connector

Provide the following configuration:

database.hostname

Example value:
123.123.123.123

database.port

1521

database.user

c##dbzuser

database.password

Example value:
1234abcdEFG

database.dbname

ORCLPDB1

topic.prefix

oracle_nickname

table.include.list

Example value:
ORCLPDB1.HOTEL

schema.history.internal.kafka.bootstrap.servers

Example values:
- Local (helm platform): platform.local:31757
- Axual-Cloud rokin cluster: bootstrap-c1-ams01-azure.axual.cloud:9094
- SAAS trial: oberon-kafka-bootstrap.axual.cloud:9094

schema.history.internal.kafka.topic

This topic name must be resolved! Use the name-resolving values acquired above.

Example resolved value:
axual-local-default-oracle_nickname._schema-changes

Supply the Axual Platform-required properties described in [_axual_platform_requirements].

Step 3 — Start the connector

Authorize the my_oracle_app source connector application to produce to and consume from all streams:

oracle_nickname._schema-changes
oracle_nickname.ORCLPDB1.HOTEL

Then start the connector application from Axual Self-Service.

Step 4 — Verify

In Axual Self-Service, use stream-browse on oracle_nickname.ORCLPDB1.HOTEL to confirm CDC events are arriving when rows are inserted or updated in Oracle.

Cleanup

When you are done:

Stop the connector application in Axual Self-Service.
Remove stream access for the application if no longer needed.

Known limitations

Oracle supplemental logging must be enabled before starting the connector.
The Oracle JDBC driver must be manually provided — it is not bundled with the connector. Use ojdbc8.jar for Java 8, ojdbc11.jar for Java 11, or ojdbc17.jar for Java 17+.
The schema.history.internal.kafka.topic value must be resolved using Axual’s name-resolving mechanism before the connector can be started.
SSL configuration values must use the keyvault: prefix format specific to Axual.
Oracle LogMiner performance can degrade under high redo log volume — consider filtering tables with table.include.list to reduce the log mining scope.

Examples

Full configuration on Axual Platform

{
  "name": "my-debezium-oracle-source",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",
    "database.hostname": "oracle.example.com",
    "database.port": "1521",
    "database.user": "c##dbzuser",
    "database.password": "Deb3zium@OraclePass",
    "database.dbname": "ORCLPDB1",
    "topic.prefix": "oracle_nickname",
    "table.include.list": "ORCLPDB1.HOTEL",
    "topic.creation.enable": "false",
    "snapshot.mode": "initial",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "schema.history.internal.kafka.bootstrap.servers": "kafka.axual.example.com:9093",
    "schema.history.internal.kafka.topic": "axual-myinstance-exampleenvironment-schema-history-oracle",
    "schema.history.internal.consumer.group.id": "my-debezium-oracle-app",
    "schema.history.internal.consumer.security.protocol": "SSL",
    "schema.history.internal.consumer.ssl.keystore.location": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.keystore.location}",
    "schema.history.internal.consumer.ssl.keystore.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.keystore.password}",
    "schema.history.internal.consumer.ssl.key.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.key.password}",
    "schema.history.internal.consumer.ssl.truststore.location": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.truststore.location}",
    "schema.history.internal.consumer.ssl.truststore.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.truststore.password}",
    "schema.history.internal.producer.security.protocol": "SSL",
    "schema.history.internal.producer.ssl.keystore.location": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.keystore.location}",
    "schema.history.internal.producer.ssl.keystore.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.keystore.password}",
    "schema.history.internal.producer.ssl.key.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.key.password}",
    "schema.history.internal.producer.ssl.truststore.location": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.truststore.location}",
    "schema.history.internal.producer.ssl.truststore.password": "${keyvault:connectors/<tenant>/<instance>/<environment>/<application>:ssl.truststore.password}",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "errors.tolerance": "none"
  }
}

License

Debezium source connector is licensed under the Apache License, Version 2.0.