# oracledb_cdc

> 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: oracledb_cdc
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: inputs/oracledb_cdc
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: inputs/oracledb_cdc.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/components/pages/inputs/oracledb_cdc.adoc
description: Enables Change Data Capture by consuming from OracleDB.
page-git-created-date: "2026-03-31"
page-git-modified-date: "2026-05-26"
---

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

**Available in:** [Cloud](https://docs.redpanda.com/cloud-data-platform/develop/connect/components/inputs/oracledb_cdc/%20%22View%20the%20Cloud%20version%20of%20this%20component%22), Self-Managed

**License**: This component requires an [enterprise license](https://docs.redpanda.com/redpanda-connect/get-started/licensing/). You can either [upgrade to an Enterprise Edition license](https://www.redpanda.com/upgrade), or [generate a trial license key](http://redpanda.com/try-enterprise) that's valid for 30 days.

Enables Change Data Capture by consuming from OracleDB.

Introduced in version 4.83.0.

Streams changes from an Oracle database for Change Data Capture (CDC). Additionally, if `stream_snapshot` is set to true, existing data in the database is also streamed.

#### Common

```yml
inputs:
  label: ""
  oracledb_cdc:
    connection_string: "" # No default (required)
    wallet_path: "" # No default (optional)
    wallet_password: "" # No default (optional)
    stream_snapshot: false
    max_parallel_snapshot_tables: 1
    snapshot_max_batch_size: 1000
    logminer:
      scn_window_size: 20000
      backoff_interval: 5s
      mining_interval: 300ms
      strategy: online_catalog
      max_transaction_events: 0
      lob_enabled: true
    include: [] # No default (required)
    exclude: [] # No default (optional)
    checkpoint_cache: "" # No default (optional)
    checkpoint_cache_table_name: RPCN.CDC_CHECKPOINT_CACHE
    checkpoint_cache_key: oracledb_cdc
    checkpoint_limit: 1024
    pdb_name: "" # No default (optional)
    auto_replay_nacks: true
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)
```

#### Advanced

```yml
inputs:
  label: ""
  oracledb_cdc:
    connection_string: "" # No default (required)
    wallet_path: "" # No default (optional)
    wallet_password: "" # No default (optional)
    stream_snapshot: false
    max_parallel_snapshot_tables: 1
    snapshot_max_batch_size: 1000
    logminer:
      scn_window_size: 20000
      backoff_interval: 5s
      mining_interval: 300ms
      strategy: online_catalog
      max_transaction_events: 0
      lob_enabled: true
    include: [] # No default (required)
    exclude: [] # No default (optional)
    checkpoint_cache: "" # No default (optional)
    checkpoint_cache_table_name: RPCN.CDC_CHECKPOINT_CACHE
    checkpoint_cache_key: oracledb_cdc
    checkpoint_limit: 1024
    pdb_name: "" # No default (optional)
    auto_replay_nacks: true
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: [] # No default (optional)
```

## [](#metadata)Metadata

This input adds the following metadata fields to each message:

-   database\_schema: The database schema for the table where the message originates from.

-   table\_name: Name of the table that the message originated from.

-   operation: Type of operation that generated the message: "read", "delete", "insert", or "update". "read" is from messages that are read in the initial snapshot phase.

-   scn: The System Change Number in Oracle.

-   schema: The table schema, for use with schema-aware downstream processors such as `schema_registry_encode`. When new columns are detected in CDC events, the schema is automatically refreshed from the Oracle catalog. Dropped columns are reflected after a connector restart.


## [](#permissions)Permissions

When using the default Oracle-based cache, the Connect user requires permission to create tables and stored procedures, and the rpcn schema must already exist. See `checkpoint_cache_table_name` for more information.

## [](#fields)Fields

### [](#auto_replay_nacks)`auto_replay_nacks`

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to `false` these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

**Type**: `bool`

**Default**: `true`

### [](#batching)`batching`

Allows you to configure a [batching policy](https://docs.redpanda.com/connect/configuration/batching/).

**Type**: `object`

```yaml
# Examples:
batching:
  byte_size: 5000
  count: 0
  period: 1s

# ---

batching:
  count: 10
  period: 1s

# ---

batching:
  check: this.contains("END BATCH")
  count: 0
  period: 1m
```

### [](#batching-byte_size)`batching.byte_size`

An amount of bytes at which the batch should be flushed. If `0` disables size based batching.

**Type**: `int`

**Default**: `0`

### [](#batching-check)`batching.check`

A [Bloblang query](https://docs.redpanda.com/connect/guides/bloblang/about/) that should return a boolean value indicating whether a message should end a batch.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
check: this.type == "end_of_transaction"
```

### [](#batching-count)`batching.count`

A number of messages at which the batch should be flushed. If `0` disables count based batching.

**Type**: `int`

**Default**: `0`

### [](#batching-period)`batching.period`

A period in which an incomplete batch should be flushed regardless of its size.

**Type**: `string`

**Default**: `""`

```yaml
# Examples:
period: 1s

# ---

period: 1m

# ---

period: 500ms
```

### [](#batching-processors)`batching.processors[]`

A list of [processors](https://docs.redpanda.com/connect/components/processors/about/) to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

**Type**: `processor`

```yaml
# Examples:
processors:
  - archive:
      format: concatenate

# ---

processors:
  - archive:
      format: lines

# ---

processors:
  - archive:
      format: json_array
```

### [](#checkpoint_cache)`checkpoint_cache`

A [cache resource](https://docs.redpanda.com/connect/components/caches/about/) to use for storing the current System Change Number (SCN) that has been successfully delivered. This allows Redpanda Connect to continue from that SCN upon restart, rather than consume the entire state of OracleDB redo logs. If not set, the default Oracle-based cache is used. See `checkpoint_cache_table_name` for more information.

**Type**: `string`

### [](#checkpoint_cache_key)`checkpoint_cache_key`

The key to use to store the snapshot position in `checkpoint_cache`. An alternative key can be provided if multiple CDC inputs share the same cache.

**Type**: `string`

**Default**: `oracledb_cdc`

### [](#checkpoint_cache_table_name)`checkpoint_cache_table_name`

The identifier for the checkpoint cache table name. If no `checkpoint_cache` field is specified, this input will automatically create a table and stored procedure under the `rpcn` schema to act as a checkpoint cache. This table stores the latest processed System Change Number (SCN) that has been successfully delivered, allowing Redpanda Connect to resume from that point upon restart rather than reconsume the entire redo log.

**Type**: `string`

**Default**: `RPCN.CDC_CHECKPOINT_CACHE`

```yaml
# Examples:
checkpoint_cache_table_name: RPCN.CHECKPOINT_CACHE
```

### [](#checkpoint_limit)`checkpoint_limit`

The maximum number of messages that can be processed at a given time. Increasing this limit enables parallel processing and batching at the output level. Any given System Change Number (SCN) will not be acknowledged unless all messages under that offset are delivered in order to preserve at least once delivery guarantees.

**Type**: `int`

**Default**: `1024`

### [](#connection_string)`connection_string`

The connection string of the Oracle database to connect to. You can supply additional connection options as URL query parameters, for example: `oracle://user:password@host:1522/service?WALLET=/opt/oracle/wallet&SSL=true`.

**Type**: `string`

```yaml
# Examples:
connection_string: oracle://username:password@host:port/service_name

# ---

connection_string: oracle://user:password@host:1522/service?WALLET=/opt/oracle/wallet&SSL=true
```

### [](#exclude)`exclude[]`

Regular expressions for tables to exclude.

**Type**: `array`

```yaml
# Examples:
exclude: SCHEMA.PRIVATETABLE
```

### [](#include)`include[]`

Regular expressions for tables to include.

**Type**: `array`

```yaml
# Examples:
include: SCHEMA.PRODUCTS
```

### [](#logminer)`logminer`

LogMiner configuration settings.

**Type**: `object`

### [](#logminer-backoff_interval)`logminer.backoff_interval`

The interval between attempts to check for new changes once all data is processed. For low traffic tables increasing this value can reduce network traffic to the server.

**Type**: `string`

**Default**: `5s`

```yaml
# Examples:
backoff_interval: 5s

# ---

backoff_interval: 1m
```

### [](#logminer-lob_enabled)`logminer.lob_enabled`

When enabled, large object (CLOB, BLOB) columns are included in both snapshot and streaming change events. When disabled, these columns are still present but contain no values. Enabling this option introduces additional performance overhead and increases memory requirements.

**Type**: `bool`

**Default**: `true`

### [](#logminer-max_transaction_events)`logminer.max_transaction_events`

The maximum number of events that can be buffered for a single transaction. If a transaction exceeds this limit it is discarded and its events will not be emitted. Set to 0 to disable the limit.

**Type**: `int`

**Default**: `0`

### [](#logminer-mining_interval)`logminer.mining_interval`

The interval between mining cycles during normal operation. Controls how frequently LogMiner polls for new changes when not caught up.

**Type**: `string`

**Default**: `300ms`

```yaml
# Examples:
mining_interval: 100ms

# ---

mining_interval: 1s
```

### [](#logminer-scn_window_size)`logminer.scn_window_size`

The SCN range to mine per cycle. Each cycle reads changes between the current SCN and current SCN + scn\_window\_size. Smaller values mean more frequent queries with lower memory usage but higher overhead; larger values reduce query frequency and improve throughput at the cost of higher memory usage per cycle.

**Type**: `int`

**Default**: `20000`

### [](#logminer-strategy)`logminer.strategy`

Controls how LogMiner retrieves data dictionary information. `online_catalog` uses the current data dictionary for best performance but cannot capture DDL changes. Currently, only `online_catalog` is supported.

**Type**: `string`

**Default**: `online_catalog`

### [](#max_parallel_snapshot_tables)`max_parallel_snapshot_tables`

Specifies a number of tables that will be processed in parallel during the snapshot processing stage.

**Type**: `int`

**Default**: `1`

### [](#pdb_name)`pdb_name`

The name of the pluggable database (PDB) to monitor. When connecting to a CDB root, LogMiner output is scoped to this PDB via SRC\_CON\_NAME filtering and catalog queries use ALTER SESSION SET CONTAINER to switch context. Requires GRANT SET CONTAINER TO <user> CONTAINER=ALL.

**Type**: `string`

### [](#snapshot_max_batch_size)`snapshot_max_batch_size`

The maximum number of rows to be streamed in a single batch when taking a snapshot.

**Type**: `int`

**Default**: `1000`

### [](#stream_snapshot)`stream_snapshot`

If set to true, the connector will query all the existing data as a part of snapshot process. Otherwise, it will start from the current System Change Number position.

**Type**: `bool`

**Default**: `false`

```yaml
# Examples:
stream_snapshot: true
```

### [](#wallet_password)`wallet_password`

Password for the `ewallet.p12` PKCS#12 wallet file. Only use this when the wallet directory contains `ewallet.p12` rather than `cwallet.sso`.

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Secrets](https://docs.redpanda.com/connect/configuration/secrets/).

> ⚠️ **CAUTION**
>
> This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see [Secrets](https://docs.redpanda.com/connect/configuration/secrets/).

**Type**: `string`

### [](#wallet_path)`wallet_path`

Path to the Oracle Wallet directory. When set, this automatically enables SSL. The directory must contain either `cwallet.sso` (auto-login, does not require a password) or `ewallet.p12` (requires `wallet_password`).

**Type**: `string`

```yaml
# Examples:
wallet_path: /opt/oracle/wallet
```