# Oracle Database CDC Patterns

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [connect-full.txt](https://docs.redpanda.com/connect-full.txt)

---
title: Oracle Database CDC Patterns
latest-connect-version: 4.93.0
latest-operator-version: v26.1.4
latest-console-tag: v3.7.3
latest-redpanda-tag: v26.1.9
docname: oracledb_cdc
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: oracledb_cdc.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/cookbooks/pages/oracledb_cdc.adoc
description: Learn how to capture, filter, transform, and route Oracle Database change data capture (CDC) events with Redpanda Connect.
page-topic-type: cookbook
personas: streaming_developer, data_engineer
learning-objective-1: Find reusable patterns for capturing Oracle Database CDC events
learning-objective-2: Look up integration patterns for routing CDC data to Redpanda and S3
learning-objective-3: Identify patterns for filtering and transforming change events
page-git-created-date: "2026-04-10"
page-git-modified-date: "2026-04-10"
---

<!-- Source: https://docs.redpanda.com/connect/cookbooks/oracledb_cdc.md -->

The Oracle Database CDC input enables capturing row-level changes from Oracle tables using LogMiner. This cookbook provides reusable patterns for filtering, transforming, and routing Oracle CDC events to Redpanda, S3, and other destinations.

Use this cookbook to:

-   Find reusable patterns for capturing Oracle Database CDC events

-   Look up integration patterns for routing CDC data to Redpanda and S3

-   Identify patterns for filtering and transforming change events


## [](#prerequisites)Prerequisites

Before using these patterns, ensure you have the following configured:

### [](#redpanda-cli)Redpanda CLI

Install the Redpanda CLI (`rpk`) to run Redpanda Connect. See [Get Started with Redpanda Connect using `rpk`](https://docs.redpanda.com/connect/get-started/quickstarts/rpk/) for installation instructions.

### [](#oracle-database-setup)Oracle Database setup

The Oracle database must be configured for CDC:

#### [](#enable-supplemental-logging)Enable supplemental logging

Supplemental logging must be enabled at the database and table level:

```sql
-- Enable database-level supplemental logging
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- Enable table-level supplemental logging (for each table)
ALTER TABLE my_schema.my_table ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
```

#### [](#enable-archivelog-mode)Enable archivelog mode

The database must be in ARCHIVELOG mode for LogMiner to access redo logs:

```sql
-- Check if archivelog is enabled
ARCHIVE LOG LIST;

-- Enable archivelog mode (if needed)
SHUTDOWN IMMEDIATE;
STARTUP MOUNT;
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;
```

#### [](#grant-permissions)Grant permissions

The database user needs specific permissions to read redo logs:

```sql
-- Create user for CDC
CREATE USER cdc_user IDENTIFIED BY <password>;

-- Grant necessary permissions
GRANT CREATE SESSION TO cdc_user;
GRANT SELECT ANY TABLE TO cdc_user;
GRANT SELECT_CATALOG_ROLE TO cdc_user;
GRANT EXECUTE_CATALOG_ROLE TO cdc_user;
GRANT SELECT ANY TRANSACTION TO cdc_user;
GRANT LOGMINING TO cdc_user;

-- Grant access to specific tables
GRANT SELECT ON my_schema.my_table TO cdc_user;

-- Create schema for checkpoint table
CREATE USER rpcn IDENTIFIED BY <password>;
GRANT CREATE TABLE TO rpcn;
GRANT CREATE PROCEDURE TO rpcn;
GRANT UNLIMITED TABLESPACE TO rpcn;
```

### [](#oracle-wallet-for-ssl-connections)Oracle Wallet (for SSL connections)

For secure connections using Oracle Wallet:

1.  Create a wallet directory:

    ```bash
    mkdir -p /opt/oracle/wallet
    ```

2.  Use Oracle Wallet Manager or `orapki` to create and configure the wallet:

    ```bash
    orapki wallet create -wallet /opt/oracle/wallet -auto_login
    orapki wallet add -wallet /opt/oracle/wallet -trusted_cert -cert ca.crt
    orapki wallet add -wallet /opt/oracle/wallet -user_cert -cert client.crt -pwd <password>
    ```

3.  The wallet directory should contain either:

    -   `cwallet.sso` (auto-login wallet, no password required)

    -   `ewallet.p12` (requires password)



### [](#environment-variables)Environment variables

The examples in this cookbook use environment variables for Oracle configuration. This allows you to keep credentials secure and separate from your pipeline configuration files.

```bash
export ORACLE_CONNECTION_STRING=oracle://cdc_user:password@host:1521/service_name (1)
export ORACLE_WALLET_PATH=/opt/oracle/wallet (2)
export ORACLE_WALLET_PASSWORD=wallet_password (3)
export REDPANDA_BROKERS=localhost:9092 (4)
export S3_BUCKET=my-cdc-bucket (5)
```

| 1 | The Oracle connection string with credentials. |
| --- | --- |
| 2 | Path to Oracle Wallet directory (optional, for SSL). |
| 3 | Wallet password if using ewallet.p12 (optional). |
| 4 | The Redpanda broker addresses (for Redpanda output examples). |
| 5 | The S3 bucket name (for S3 output examples). |

## [](#capture-cdc-events)Capture CDC events

The simplest pattern captures all change events from Oracle tables and outputs them with metadata:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"
      - "MY_SCHEMA\\.CUSTOMERS"
    exclude:
      - "MY_SCHEMA\\..*_TEMP"

pipeline:
  processors:
    # Extract the change event details
    - mapping: |
        root = this
        root.metadata.operation = meta("operation")
        root.metadata.table = meta("table_name")
        root.metadata.schema = meta("database_schema")
        root.metadata.scn = meta("scn")
        root.metadata.timestamp = now()

output:
  stdout:
    codec: lines
```

For details on the CDC event message structure and available metadata fields, see the [metadata section](https://docs.redpanda.com/connect/components/inputs/oracledb_cdc/#_metadata) in the connector reference.

## [](#filter-cdc-events)Filter CDC events

You can filter events to process only specific change types:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"

pipeline:
  processors:
    # Filter out delete events
    - mapping: |
        root = if meta("operation") == "delete" { deleted() } else { this }

    # Add operation metadata to the message
    - mapping: |
        root = this.merge({
          "operation": meta("operation"),
          "table": meta("table_name")
        })

output:
  stdout:
    codec: lines
```

This example:

-   Filters out `delete` events using `deleted()`

-   Adds operation type and timestamp to the output

-   Transforms the event to include metadata


## [](#stream-initial-snapshot)Stream initial snapshot

Capture both existing data and ongoing changes:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"
    stream_snapshot: true (1)
    snapshot_max_batch_size: 5000 (2)
    max_parallel_snapshot_tables: 2 (3)

pipeline:
  processors:
    - mapping: |
        root = this

        # Tag snapshot vs CDC events
        root.event_type = if meta("operation") == "read" {
          "snapshot"
        } else {
          "cdc"
        }

output:
  stdout:
    codec: lines
```

| 1 | Enable initial snapshot of existing table data. |
| --- | --- |
| 2 | Number of rows per batch during snapshot (higher = faster, more memory). |
| 3 | Number of tables to snapshot in parallel. |

> 📝 **NOTE**
>
> Snapshot events have `operation` metadata set to `"read"`, while CDC events use `"insert"`, `"update"`, or `"delete"`.

## [](#route-to-redpanda)Route to Redpanda

Stream Oracle changes to Redpanda for real-time processing:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"
      - "MY_SCHEMA\\.CUSTOMERS"

pipeline:
  processors:
    # Create a composite key from the primary key fields
    - mapping: |
        meta kafka_key = this.ID.string()

        # Add CDC metadata as message headers
        meta operation = meta("operation")
        meta table = meta("table_name")
        meta scn = meta("scn")

output:
  redpanda:
    seed_brokers: ["${REDPANDA_BROKERS}"]
    topic: 'oracle-cdc-${! meta("table_name").lowercase() }' (1)
    max_in_flight: 10
    compression: snappy
    batching:
      count: 100
      period: 100ms
```

| 1 | Route to different topics based on table name. |
| --- | --- |

This example:

-   Creates a message key from the primary key for consistent partitioning

-   Routes each table to a separate topic

-   Batches messages for efficient delivery

-   Adds CDC metadata as Kafka headers


## [](#route-to-s3)Route to S3

Archive CDC events to S3 for long-term storage and analytics:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"

pipeline:
  processors:
    - mapping: |
        root = this
        root.operation = meta("operation")
        root.table = meta("table_name")
        root.scn = meta("scn")
        root.captured_at = now().ts_format("2006-01-02T15:04:05Z07:00")

output:
  aws_s3:
    bucket: ${S3_BUCKET}
    path: 'oracle-cdc/${! meta("table_name") }/${! timestamp_unix().ts_format("2006/01/02/15") }/${! uuid_v4() }.jsonl' (1)
    batching:
      count: 1000
      period: 5m
      processors:
        - archive:
            format: json_array (2)
        - compress:
            algorithm: gzip (3)
```

| 1 | Organize files by table and time-based partitions (year/month/day/hour). |
| --- | --- |
| 2 | Archive as JSON array for easier downstream parsing. |
| 3 | Compress with gzip to reduce storage costs. |

This example:

-   Organizes files by table name and hourly partitions

-   Batches events and archives them as compressed JSON

-   Uses UUID file names to prevent collisions


## [](#route-by-event-type)Route by event type

Route different event types to different destinations:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"

pipeline:
  processors:
    # Add routing metadata
    - mapping: |
        root = this
        root.event_type = meta("operation")
        root.table = meta("table_name")

output:
  switch:
    cases:
      # Route INSERT events
      - check: this.event_type == "insert"
        output:
          redpanda:
            seed_brokers: ["${REDPANDA_BROKERS}"]
            topic: orders-inserts

      # Route UPDATE events
      - check: this.event_type == "update"
        output:
          redpanda:
            seed_brokers: ["${REDPANDA_BROKERS}"]
            topic: orders-updates

      # Route DELETE events
      - check: this.event_type == "delete"
        output:
          redpanda:
            seed_brokers: ["${REDPANDA_BROKERS}"]
            topic: orders-deletes

      # Fallback for snapshot reads
      - output:
          drop: {}
```

This pattern:

-   Separates processing pipelines for inserts, updates, and deletes

-   Routes each operation type to a dedicated topic

-   Enables specialized downstream consumers per operation type


## [](#detect-changed-columns)Detect changed columns

For update operations, identify which columns changed:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"

processors:
  - mapping: |
      # Only process update operations
      root = if meta("operation") != "update" { deleted() }

      # Extract the update event details
      # Oracle CDC provides before/after values for updates
      root.id = this.ID
      root.table = meta("table_name")
      root.operation = meta("operation")
      root.scn = meta("scn")

      # Include the full after-state
      root.after = this

output:
  stdout:
    codec: lines
```

> 📝 **NOTE**
>
> Oracle CDC captures the full row state after the update. For fine-grained change detection, you may need to maintain a previous state externally or use Oracle’s built-in change tracking features.

## [](#checkpointing)Checkpointing

The Oracle CDC input automatically manages checkpoints using a table in the Oracle database:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"
    checkpoint_cache_table_name: RPCN.CDC_CHECKPOINT_CACHE (1)
    checkpoint_cache_key: orders_cdc_job (2)
    checkpoint_limit: 500 (3)
```

| 1 | Table name for storing checkpoint state (created automatically if it doesn’t exist). |
| --- | --- |
| 2 | Unique key to identify this CDC job (useful when running multiple CDC pipelines). |
| 3 | Checkpoint after every 500 messages (lower = better recovery, higher = fewer writes). |

The checkpoint table stores the System Change Number (SCN) of the last successfully processed event. On restart, CDC resumes from the last checkpoint.

### [](#using-an-external-cache)Using an external cache

For better separation or to use a different database for checkpointing:

```yaml
cache_resources:
  - label: postgres_cache
    postgres:
      dsn: postgres://user:pass@postgres-host:5432/checkpoints

input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\.ORDERS"
    checkpoint_cache: postgres_cache (1)
    checkpoint_cache_key: orders_cdc_job
```

| 1 | Use an external cache resource instead of the default Oracle table. |
| --- | --- |

## [](#performance-tuning)Performance tuning

Optimize throughput and latency with these LogMiner settings:

```yaml
input:
  oracledb_cdc:
    connection_string: ${ORACLE_CONNECTION_STRING}
    include:
      - "MY_SCHEMA\\..*"
    logminer:
      scn_window_size: 50000 (1)
      mining_interval: 100ms (2)
      backoff_interval: 1s (3)
      max_transaction_events: 10000 (4)
      lob_enabled: true (5)
    checkpoint_limit: 2000 (6)
    batching:
      count: 100
      period: 1s
```

| 1 | Number of SCNs to process in each LogMiner session (higher = more memory, fewer sessions). |
| --- | --- |
| 2 | How often to query LogMiner for new changes (lower = less latency, more CPU). |
| 3 | Backoff when no changes are detected (reduces CPU usage during idle periods). |
| 4 | Maximum events to buffer per transaction (0 = unlimited). |
| 5 | Enable capture of LOB (CLOB/BLOB) columns. |
| 6 | Number of messages to process before checkpointing. |

### [](#throughput-considerations)Throughput considerations

-   Larger `scn_window_size` reduces LogMiner overhead but increases memory usage

-   Shorter `mining_interval` reduces latency but increases database load

-   Higher `checkpoint_limit` improves throughput but increases recovery time after failures

-   LOB columns significantly impact performance; disable `lob_enabled` if not needed


## [](#secure-connections-with-oracle-wallet)Secure connections with Oracle Wallet

Use Oracle Wallet for SSL/TLS connections:

```yaml
input:
  oracledb_cdc:
    connection_string: oracle://cdc_user:password@host:1522/service
    wallet_path: ${ORACLE_WALLET_PATH} (1)
    wallet_password: ${ORACLE_WALLET_PASSWORD} (2)
    include:
      - "MY_SCHEMA\\.ORDERS"
```

| 1 | Path to the wallet directory containing cwallet.sso or ewallet.p12. |
| --- | --- |
| 2 | Wallet password (only required for ewallet.p12 files). |

Alternatively, specify wallet configuration in the connection string:

```yaml
input:
  oracledb_cdc:
    connection_string: oracle://user:password@host:1522/service?WALLET=/opt/oracle/wallet&SSL=true
    include:
      - "MY_SCHEMA\\.ORDERS"
```

## [](#troubleshoot)Troubleshoot

### [](#no-events-received)No events received

If you’re not receiving events:

1.  Verify supplemental logging is enabled:

    ```sql
    -- Check database-level logging
    SELECT SUPPLEMENTAL_LOG_DATA_MIN FROM V$DATABASE;

    -- Check table-level logging
    SELECT LOG_GROUP_NAME, TABLE_NAME, DECODE(ALWAYS, 'ALWAYS', 'Unconditional', NULL, 'Conditional') ALWAYS
    FROM DBA_LOG_GROUPS
    WHERE OWNER = 'MY_SCHEMA' AND TABLE_NAME = 'MY_TABLE';
    ```

2.  Check that the database is in ARCHIVELOG mode:

    ```sql
    SELECT LOG_MODE FROM V$DATABASE;
    ```

3.  Verify the user has correct permissions:

    ```sql
    SELECT * FROM DBA_SYS_PRIVS WHERE GRANTEE = 'CDC_USER';
    ```

4.  Make sure changes are being made to the monitored tables


### [](#slow-event-processing)Slow event processing

If CDC is lagging behind:

-   Increase `scn_window_size` to process more changes per LogMiner session

-   Reduce `checkpoint_limit` to checkpoint more frequently (helps during restarts)

-   Disable `lob_enabled` if LOB columns aren’t needed

-   Check Oracle alert logs for performance issues

-   Monitor the size of redo logs and archive logs


### [](#connection-issues-with-oracle-wallet)Connection issues with Oracle Wallet

If SSL connections fail:

1.  Verify wallet files exist and have correct permissions:

    ```bash
    ls -la /opt/oracle/wallet
    ```

2.  Check Oracle TNS configuration

3.  Verify the wallet password for `ewallet.p12` files

4.  Test the connection using Oracle tools (sqlplus, SQL Developer)


### [](#checkpoint-table-errors)Checkpoint table errors

If you see errors related to the checkpoint table:

1.  Ensure the `rpcn` schema exists:

    ```sql
    CREATE USER rpcn IDENTIFIED BY <password>;
    GRANT CREATE TABLE TO rpcn;
    GRANT CREATE PROCEDURE TO rpcn;
    GRANT UNLIMITED TABLESPACE TO rpcn;
    ```

2.  Verify the CDC user has access to the checkpoint table:

    ```sql
    GRANT SELECT, INSERT, UPDATE ON rpcn.cdc_checkpoint_cache TO cdc_user;
    ```


## [](#suggested-reading)Suggested reading

-   [Oracle CDC Input Reference](https://docs.redpanda.com/connect/components/inputs/oracledb_cdc/)

-   [Redpanda Output](https://docs.redpanda.com/connect/components/outputs/redpanda/)

-   [AWS S3 Output](https://docs.redpanda.com/connect/components/outputs/aws_s3/)

-   [Oracle LogMiner Documentation](https://docs.oracle.com/en/database/oracle/oracle-database/19/sutil/oracle-logminer-utility.html)

-   [Configuring Supplemental Logging](https://docs.oracle.com/en/database/oracle/oracle-database/19/sutil/configuring-supplemental-logging.html)


## Suggested labs

-   [Stream Jira Issues to Redpanda for Real-Time Metrics](https://docs.redpanda.com/labs/docker-compose/jira-metrics-pipeline/)

[Search all labs](https://docs.redpanda.com/labs)