<a id="connect-oracle-xstream-cdc-source-config"></a>

# Configuration Reference for Oracle XStream CDC Source Connector for Confluent Platform

The sections in this page define each of the Oracle XStream CDC Source connector
configuration properties.

#### NOTE
These are properties for the self-managed connector. If you are using
Confluent Cloud, see [Oracle XStream CDC Source Connector for Confluent Cloud](/cloud/current/connectors/cc-oracle-xstream-cdc-source/cc-oracle-xstream-cdc-source-features.html).

## Connector

`name`
: Unique name for the connector.
  <br/>
  * Type: string
  * Valid Values: A string at most 64 characters long
  * Importance: high

`connector.class`
: Java class name for the connector. It is always set to `io.confluent.connect.oracle.xstream.cdc.OracleXStreamSourceConnector`.
  <br/>
  * Type: string
  * Default: NA
  * Importance: high

`tasks.max`
: The maximum number of tasks for this connector. This configuration does not apply to the Oracle connector because it always operates with a single task.
  <br/>
  * Type: int
  * Default: 1
  * Valid Values: [1,…,1]
  * Importance: high

### Database

`database.hostname`
: IP address or hostname of the Oracle database server.
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^[a-zA-Z0-9-_.]+$`
  * Importance: high

`database.port`
: Port number of the Oracle database server.
  <br/>
  * Type: int
  * Default: 1521
  * Valid Values: [1,…,65535]
  * Importance: high

`database.user`
: Name of the Oracle database user to use when connecting to the database.
  <br/>
  * Type: string
  * Importance: high

`database.password`
: Password of the Oracle database user to use when connecting to the database.
  <br/>
  * Type: password
  * Importance: high

`database.dbname`
: Name of the database to connect to. In a multitenant architecture, this is the name of the container database (CDB).
  <br/>
  * Type: string
  * Importance: high

`database.service.name`
: Name of the database service to which to connect. In a multitenant container database, this is the service used to connect to the CDB. For Oracle Real Application Clusters (RACs), use the service created by Oracle XStream.
  <br/>
  * Type: string
  * Importance: high

`database.pdb.name`
: Name of the pluggable database to connect to in a multitenant architecture. The container database (CDB) name must be given via `database.dbname` in this case. This configuration should not be specified when connecting to a non-container database.
  <br/>
  * Type: string
  * Importance: high

`database.out.server.name`
: Name of the XStream outbound server to connect to.
  <br/>
  * Type: string
  * Importance: high

`database.os.timezone`
: Specifies the database server’s operating system timezone. This is used to read the time when the LCR was generated at the source database. The default timezone is UTC. The value has to be a valid `java.time.ZoneId` identifier.
  <br/>
  * Type: string
  * Default: UTC
  * Importance: low

`database.processor.licenses`
: Specifies the number of Oracle processor licenses required for the source database server or cluster. The is determined by multiplying the total number of processor cores by a core processor licensing factor, as specified in the Oracle Processor Core Factor Table.
  <br/>
  * Type: int
  * Default: NA
  * Importance: medium

`database.tls.mode`
: Specifies whether to use Transport Layer Security (TLS) to connect to the Oracle database.
  <br/>
  * Type: string
  * Default: disable
  * Valid Values: disable, one-way, two-way
  * Importance: medium
  <br/>
  You can set one of the following values:
  <br/>
  * **disable** (default): Does not use a TLS connection.
  * **one-way**: Uses a TLS-encrypted connection and also verifies the server’s TLS certificate.
    The trusted CA root certificate used to sign the server’s certificate must be present either
    in the default certificate store of the system running the connector or stored within a
    client wallet provided to the connector using the `database.wallet.location` configuration property.
  * **two-way**: Uses a TLS-encrypted connection and also verifies the server’s TLS certificate.
    In addition, the client presents its own certificate, which the server verifies to authenticate
    the client (mutual TLS). The client certificate and the trusted CA root certificate used to
    sign the server’s certificate must be stored within a client wallet and provided to the
    connector using the `database.wallet.location` configuration property.

`database.wallet.location`
: Specifies the directory path containing the Oracle client wallet (`cwallet.sso`) that contains the certificates used
  for TLS connections between the connector and the database server. This must be a Single Sign-On (SSO) auto-login
  wallet. For example, if your wallet file is located at `/opt/oracle/wallet/cwallet.sso`, set this to `/opt/oracle/wallet`.
  <br/>
  * Type: string
  * Importance: medium
  <br/>
  Note the following:
  <br/>
  * For one-way TLS: A client wallet is required only if the system’s default certificate store does not contain the
    trusted CA root certificate used to sign the server’s certificate. In that case, the client wallet must include
    the trusted CA root certificate.
  * For two-way TLS (mutual TLS): A client wallet is required and must contain both
    the client certificate and the trusted CA root certificate used to sign the server’s certificate.

## Topic name

`topic.prefix`
: Topic prefix that provides a namespace for the Oracle database server or cluster from which the connector captures changes. The topic prefix should be unique across all other connectors, since it is used as a prefix for all Kafka topic names that receive events from this connector. Only alphanumeric characters, hyphens, dots and underscores are accepted.
  <br/>
  * Type: string
  * Importance: high

#### WARNING
Do not change the value of this property. If you change the value, after a restart, instead of continuing to emit events to the original topics, the connector emits subsequent events to topics whose names are based on the new value. The connector is also unable to recover its database schema history topic.

<a id="connect-oracle-xstream-cdc-source-connector-config"></a>

## Connector configuration

`table.include.list`
: An optional, comma-separated list of regular expressions that match fully-qualified table identifiers for the tables whose changes you want to capture. When this property is set, the connector will only capture changes from the specified tables. Each identifier is of the form `schemaName.tableName`. By default, the connector captures changes from all non-system tables in each captured database.
  To match the name of a table, the connector applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire identifier for the table; it does not match substrings that might be present in a table name.
  If you include this property in the configuration, do not set the `table.exclude.list` property.
  <br/>
  * Type: string
  * Importance: high

`table.exclude.list`
: An optional, comma-separated list of regular expressions that match fully-qualified table identifiers for the tables whose changes you do not want to capture. When this property is set, the connector captures changes from any table that is not specified in the exclude list. Each identifier is of the form `schemaName.tableName`.
  To match the name of a table, the connector applies the regular expression that you specify as an anchored regular expression. That is, the specified expression is matched against the entire identifier for the table; it does not match substrings that might be present in a table name.
  If you include this property in the configuration, do not set the `table.include.list` property.
  <br/>
  * Type: string
  * Importance: medium

`column.include.list`
: An optional, comma-separated list of regular expressions that match
  fully-qualified column identifiers to be included in change event values. Each identifier is
  of the form `schemaName.tableName.columnName`. To match the name of a column, the connector applies
  the regular expression that you specify as an anchored regular expression. That is, the specified
  expression is matched against the entire identifier for the column; it does not match substrings
  that might be present in a column name. If you use this property, do not
  set the `column.exclude.list` property.
  <br/>
  * Type: string
  * Importance: medium
  <br/>
  #### NOTE
  Primary key columns are always included in an event’s key, even if you do not use
  this property to explicitly include its value.

`column.exclude.list`
: An optional, comma-separated list of regular expressions that match
  fully-qualified column identifiers to be excluded from change event values. Each identifier is
  of the form `schemaName.tableName.columnName`. To match the name of a column, the connector
  applies the regular expression that you specify as an anchored regular expression. That is, the
  specified expression is matched against the entire identifier for the column; it does not match
  substrings that might be present in a column name. If you use this property,
  do not set the `column.include.list` property.
  <br/>
  * Type: string
  * Importance: medium
  <br/>
  #### NOTE
  Primary key columns are always included in an event’s key, even if you
  use this property to explicitly exclude its value.

`snapshot.mode`
: The criteria for running a snapshot upon startup of the connector.
  <br/>
  * Type: string
  * Default: `initial`
  * Valid values: `initial`, `no_data` and `recovery`
  * Importance: medium
  <br/>
  You can set one of the following values:
  <br/>
  * **initial** (default): The snapshot includes both the structure (schema) and data of the captured tables. Specify this value to populate topics with a complete representation of the data from the captured tables. After the snapshot completes, the connector begins to stream event records for subsequent database changes.
  * **no_data**: The snapshot includes only the structure (schema) of captured tables. Specify this value if you want the connector to capture data only for changes that occur after the snapshot. After the snapshot completes, the connector begins to stream event records for subsequent database changes.
  * **recovery**: Set this option to restore a database schema history topic that is lost or corrupted. After a restart, the connector runs a snapshot that rebuilds the topic from the source tables. You can also set the property to periodically prune a database schema history topic that experiences unexpected growth.
  <br/>
  #### WARNING
  Do not use recovery mode to perform a snapshot if schema changes were committed to the database after the last connector shutdown.

`snapshot.fetch.size`
: The number of rows to provide as a hint to the JDBC driver that should be fetched from the database when more rows are needed when taking a snapshot. A value of ‘0’ uses the default JDBC fetch size.
  <br/>
  * Type: int
  * Default: 10,000
  * Importance: medium

`snapshot.max.threads`
: Specifies the number of threads the connector uses during an initial snapshot. To enable parallel processing of the initial snapshot, set this property to a value greater than 1. With parallel snapshots, the connector processes multiple tables at the same time.
  <br/>
  * Type: int
  * Default: 1
  * Importance: medium

`snapshot.database.errors.max.retries`
: Specifies the number of retry attempts the connector will make to snapshot a table if a database error occurs. This configuration property currently only retries failures related to `ORA-01466` error. By default, no retries are attempted.
  <br/>
  * Type: int
  * Default: 0
  * Valid Values: [0,…,3]
  * Importance: Low

`snapshot.select.statement.overrides`
: An optional, comma-separated list of tables for which the connector uses custom
  SELECT statements to determine which rows to include in the initial snapshot.
  The connector runs these statements during the initial snapshot instead of
  selecting all rows. It does not apply to events that the connector reads from
  the log during the streaming phase.
  <br/>
  * Type: password
  * Importance: medium
  <br/>
  This property works together with secondary per-table properties:
  <br/>
  * **Main property** (`snapshot.select.statement.overrides`): A comma-separated list of fully-qualified table identifiers in the format `schemaName.tableName`. This list identifies the tables that have custom snapshot SELECT statements.
  * **Per-table properties**: For each table listed in the main property, define a property in the format `snapshot.select.statement.overrides.schemaName.tableName` that specifies the custom SELECT statement to run during the initial snapshot. If a table is listed in the main property but its secondary property is missing, the connector logs a warning and uses the default snapshot behavior for that table.
  <br/>
  The case of object names in the fully-qualified name must match exactly how they are stored in the Oracle database. Unquoted identifiers are case-insensitive and are treated as uppercase, whereas quoted identifiers are case-sensitive and must be enclosed in double quotes. If any of the object names contain special characters such as spaces or periods (`.`), enclose them in double quotes.
  <br/>
  #### WARNING
  If you override the snapshot SELECT statement, any row changes that occur between the snapshot’s start SCN and when the override query actually runs might be captured twice: once during the snapshot, and again when streaming later replays that same window.

`decimal.handling.mode`
: Specifies how the connector should handle `NUMBER`, `DECIMAL` and `NUMERIC` columns.
  <br/>
  * Type: string
  * Default: precise
  * Valid Values: double, precise, string
  * Importance: medium
  <br/>
  You can set one of the following values:
  <br/>
  * **precise** (default): Uses java.math.BigDecimal to represent values, which are encoded in the change events using a binary representation and Kafka Connect’s org.apache.kafka.connect.data.Decimal type. Depending on the precision and scale, the most appropriate Kafka Connect integer type is used for integral values, ensuring that the value is represented without any loss of precision.
  * **string**: Encodes values as formatted strings. Using the string option is easier to consume, but results in a loss of semantic information about the real type.
  * **double**: Represents values using Java’s double. Using double values is easier, but can result in a loss of precision.

`binary.handling.mode`
: Specifies how the connector should handle binary (BLOB) columns.
  <br/>
  * Type: string
  * Default: bytes
  * Valid Values: bytes, base64, base64-url-safe, hex
  * Importance: medium
  <br/>
  You can set one of the following values:
  <br/>
  * **bytes** (default): Represents binary data as byte array.
  * **base64**: Represents binary data as base64-encoded string.
  * **base64-url-safe**: Represents binary data as base64-url-safe-encoded string.
  * **hex**: Represents binary data as hex-encoded (base16) string.

`time.precision.mode`
: Specifies how the connector should handle time, date, and timestamps columns.
  <br/>
  * Type: string
  * Default: adaptive
  * Valid Values: adaptive, connect
  * Importance: medium
  <br/>
  You can set one of the following options:
  <br/>
  * **adaptive** (default): Bases the precision of time, date, and timestamp values on the database column’s precision.
  * **connect**: always represents time, date, and timestamp values using Kafka Connect’s built-in representations for Time, Date, and Timestamp, which uses millisecond precision regardless of the database columns’ precision.

`query.fetch.size`
: The number of rows to provide as a hint to the JDBC driver that should be fetched from the database when more rows are needed. A value of ‘0’ uses the default JDBC fetch size.
  <br/>
  * Type: int
  * Default: 10,000
  * Importance: medium

`tombstones.on.delete`
: Controls whether a delete event is followed by a tombstone event. After a source record is deleted, a tombstone event (the default behavior) enables Kafka to completely delete all events that share the key of the deleted row in topics that have log compaction enabled.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium
  <br/>
  The following values are possible:
  <br/>
  * **true**: For each delete operation, the connector emits a delete event and a subsequent tombstone event.
  * **false**: For each delete operation, the connector emits only a delete event.

`skipped.operations`
: A comma-separated list of the operation types to skip during streaming.
  <br/>
  * Type: string
  * Default: t
  * Importance: low
  <br/>
  You can configure the connector to skip the following types of operations:
  <br/>
  * c (create/insert)
  * u (update)
  * d (delete)
  * t (truncate)
  <br/>
  You use `none` to indicate that no operations are skipped. By default, only truncate (`t`) operations are skipped.

`schema.name.adjustment.mode`
: Specifies how schema names should be adjusted for compatibility with the message converter used by the connector
  <br/>
  * Type: string
  * Default: none
  * Valid Values: avro, avro_unicode, none
  * Importance: low
  <br/>
  The following values are possible:
  <br/>
  * **none** (default): does not apply any adjustment.
  * **avro**: replaces the characters that cannot be used in the Avro type name with underscore.
  * **avro_unicode**: replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like \_uxxxx. Note: \_ is an escape sequence like backslash in Java.

`field.name.adjustment.mode`
: Specifies how field names should be adjusted for compatibility with the message converter used by the connector.
  <br/>
  * Type: string
  * Default: none
  * Valid Values: avro, avro_unicode, none
  * Importance: low
  <br/>
  The following values are possible:
  <br/>
  * **none** (the default): does not apply any adjustment.
  * **avro**: replaces the characters that cannot be used in the Avro type name with underscore.
  * **avro_unicode**: replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like \_uxxxx.
  <br/>
  #### NOTE
  \_ is an escape sequence like backslash in Java.

`schema.history.internal.kafka.topic`
: The name of the topic for the database schema history. A new topic with provided name is created if it doesn’t already exist. If the topic already exists, ensure that is it has a single partition, infinite retention period, and is not in use by any other connector.
  <br/>
  * Type: string
  * Default: NA
  * Importance: high

`schema.history.internal.kafka.bootstrap.servers`
: A list of host/port pairs that the connector will use for establishing the initial connection to the Kafka cluster for retrieving database schema history previously stored by the connector. This should point to the same Kafka cluster used by the Kafka Connect process.
  <br/>
  * Type: string
  * Default: NA
  * Importance: high

`schema.history.internal.skip.unparseable.ddl`
: A boolean value that specifies whether the connector should ignore a DDL statement that cannot be parsed or stop processing for a human to address the issue. The safe default is `false` which causes the connector to fail when it encounters an unparseable DDL statement. Setting the value to `true` should be done with care as it will cause the connector to skip processing any DDL statement it cannot parse, and this could potentially lead to schema mismatches and data loss.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`topic.heartbeat.prefix`
: Controls the name of the topic to which the connector sends heartbeat messages. The heartbeat topic name follows this format: `${topic.heartbeat.prefix}.{topic.prefix}`.
  <br/>
  * Type: string
  * Default: `__cflt-oracle-heartbeat`
  * Importance: low

`heartbeat.interval.ms`
: Controls how often the connector sends heartbeat messages to a heartbeat topic. It is useful in situations when no changes occur in the captured tables for an extended period. In such cases, there are no change event messages generated, causing the committed source offset to remain unchanged. As a result, the connector is unable to update the processed low watermark on the outbound server which could result in the database retaining archived redo log files longer than needed. The default value is 0 which disables the heartbeat mechanism.
  <br/>
  * Type: int
  * Default: 0
  * Valid Values: [0,…]
  * Importance: medium

`unavailable.value.placeholder`
: Specifies the constant that the connector uses to indicate that the
  original value is unavailable and not provided by the database.
  <br/>
  * Type: string
  * Default: \_\_cflt_unavailable_value
  * Importance: medium

`lob.oversize.threshold`
: Specifies the maximum size threshold (in bytes) for large object (LOB)
  column values, including CLOB, NCLOB, and BLOB. For CLOB and NCLOB values, the connector calculates the size as
  the UTF-8 encoded byte length of the string. If a LOB value exceeds this threshold, the connector handles it according
  to the strategy specified using the `lob.oversize.handling.mode` configuration. The default value is -1, which
  disables oversize handling.
  <br/>
  * Type: int
  * Default: -1
  * Importance: low

`lob.oversize.handling.mode`
: Defines how the connector handles LOB column values that exceed the
  size threshold specified using the `lob.oversize.threshold` configuration.
  <br/>
  * Type: string
  * Default: fail
  * Valid Values: fail, skip
  * Importance: low
  <br/>
  Select one of the following options:
  <br/>
  * **fail** (default): The connector stops processing and reports an error.
  * **skip**: The connector replaces the LOB value with a placeholder specified using the `skip.value.placeholder` configuration.

`skip.value.placeholder`
: Specifies the constant provided by the connector to indicate that the original value
  was skipped by the connector due to exceeding the configured size threshold.
  <br/>
  * Type: string
  * Default: \_\_cflt_skipped_value
  * Importance: low

`signal.data.collection`
: Specifies the fully-qualified name of the signaling table, using the
  format: `<databaseName>.<schemaName>.<tableName>`. This table is used to send signals
  to the connector when using the source channel. Source signaling is disabled if you do not set this property.
  <br/>
  The case of object names in the fully-qualified name must exactly match how they are stored
  in the Oracle database. Unquoted identifiers are case-insensitive and are treated as uppercase
  by default, whereas quoted identifiers are case-sensitive.
  <br/>
  When connecting to a Pluggable Database (PDB), use the name of the PDB as the database name
  component of the fully-qualified name.
  <br/>
  * Type: string
  * Importance: medium

`signal.enabled.channels`
: Specifies which signaling channels are enabled for the connector.
  <br/>
  Supported values are:
  <br/>
  * **source** (default): Signals are read from a signaling table in the source database.
  * **kafka**: Signals are consumed from a Kafka topic.
  <br/>
  You can enable multiple channels by providing a comma-separated list. If not set, the
  connector enables only the `source` channel by default.
  <br/>
  * Type: string
  * Default: source
  * Valid Values: source, kafka
  * Importance: medium

`signal.kafka.topic`
: Specifies the name of the Kafka topic the connector monitors for
  signals. The topic must have a single partition to ensure signal ordering is preserved.
  <br/>
  * Type: string
  * Importance: medium

`signal.kafka.bootstrap.servers`
: A list of host/port pairs that the connector uses to establish the initial
  connection to the Kafka cluster for retrieving signals to the connector. This must
  point to the same Kafka cluster used by the Kafka Connect process.
  <br/>
  * Type: string
  * Importance: medium

`outbound.server.attach.position.mode`
: Specifies the position the connector passes when attaching to the XStream
  outbound server.
  <br/>
  * Type: string
  * Default: connector_managed
  * Valid Values: connector_managed, server_managed
  * Importance: low
  <br/>
  Select one of the following options:
  <br/>
  * **connector_managed** (default): The connector passes the last position it
    persisted to the Kafka Connect offsets topic. The outbound server resumes
    streaming from the first LCR with a position greater than this value. If no
    offset is stored, the connector passes the position corresponding to the
    database’s current SCN (System Change Number).
  * **server_managed**: The connector passes `NULL` to the outbound server
    instead of the stored position. The outbound server automatically determines
    the processed low position and starts streaming from the first LCR with a
    position greater than this value. The connector then discards any LCR with a
    position less than or equal to the last position it persisted to the
    Kafka Connect offsets topic.
  <br/>
  #### WARNING
  In this mode, resetting the connector’s stored offset to a position below
  the outbound server’s processed low position does not cause the connector to
  fail, unlike the behavior in `connector_managed` mode. The reset has no
  effect, and the connector resumes streaming from the outbound server’s
  processed low position.
  <br/>
  Before changing the default value, contact Confluent Support.

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

## CSFLE and CSPE configurations

`csfle.enabled`

Accepts a boolean value. CSFLE is enabled for the connector if `csfle.enabled` is set to True.

* Type: boolean
* Default: False

`auto.register.schemas`

Specifies if the Serializer should attempt to register the Schema with Schema Registry.

* Type: boolean
* Default: true
* Importance: medium

`use.latest.version`

Only applies when `auto.register.schemas` is set to false. If `auto.register.schemas` is set to false and `use.latest.version` is set to true, then instead of deriving a schema for the object passed to the client for serialization, Schema Registry uses the latest version of the schema in the subject for serialization.

* Type: boolean
* Default: true
* Importance: medium