<a id="cc-oracle-xstream-cdc-source-configure-connector"></a>

# Oracle XStream CDC Source Connector for Confluent Cloud

The fully managed Oracle XStream Change data capture (CDC) Source connector
for Confluent Cloud captures
every `INSERT`, `UPDATE`, and `DELETE` on Oracle database tables and publishes them
as change event records to Apache Kafka® topics, using Oracle XStream Out internally.
You can include or exclude specific tables with regular expressions on the
table identifiers.

The Confluent Cloud connector plugin name is `OracleXStreamSource`. The underlying
connector class is
`io.confluent.connect.oracle.xstream.cdc.OracleXStreamSourceConnector`.

This Quick Start is for the fully managed Confluent Cloud connector. If you are installing
the connector locally for Confluent Platform,
see [Oracle XStream CDC Source](https://docs.confluent.io/kafka-connectors/oracle-xstream-cdc-source/current/) for Confluent Platform.

If you require private networking for fully managed connectors, make sure to set up the proper
networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](../networking/internet-resource.md#clusters-connect-cloud).

<a id="cc-oracle-xstream-cdc-source-supported-versions"></a>

## Supported Versions

Be sure to review the following information before using the Oracle XStream CDC Source connector.

### Oracle versions

The connector is compatible with the following Oracle versions:

- Oracle 19c Enterprise Edition
- Oracle 19c Standard Edition
- Oracle 21c Enterprise Edition
- Oracle 21c Standard Edition

The connector supports Oracle Exadata.

The connector supports Oracle Database 19c using the non-CDB architecture
on Amazon RDS for Oracle. For more information, see [Working with Amazon RDS for Oracle](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs-check-amazon-rds).

### Java versions

The connector requires Java version 17 or higher.

## Limitations

Be sure to review the following information:

* The connector has not been tested against managed database services from cloud
  service providers (CSPs), other than Amazon RDS for Oracle.
* The connector does not work with Oracle Autonomous Databases.
* The connector cannot run XStream capture on an Oracle Data Guard standby
  database; the XStream capture process and outbound server require a
  primary. To capture changes from a database protected by Data Guard, see
  [Cascading Downstream Capture for Oracle XStream CDC Source Connector for Confluent Cloud](cascading-downstream-capture.md#xstream-cascading-downstream-capture), where an Active Data Guard
  standby serves as the redo-transport hop and can also serve the
  connector’s initial snapshot and schema reads.
* If you plan to use one or more Single Message Transforms (SMTs), see [SMT Limitations](../single-message-transforms.md#cc-single-message-transforms-limitations).
* The connector does not support the following Single Message Transforms
  (SMTs): [TimestampRouter](../transforms/timestamprouter.md#timestamprouter) and
  [MessageTimestampRouter](../transforms/messagetimestamprouter.md#messagetimestamprouter).

<a id="cc-oracle-xstream-cdc-source-custom-offsets"></a>

## Manage custom offsets

You can manage the offsets for this connector. Offsets provide information on the
point in the system from which the connector is accessing data. For more
information, see [Manage Offsets for Fully Managed Connectors in Confluent Cloud](../offsets.md#connect-custom-offsets).

**To manage offsets**:

- Use Confluent Cloud APIs. For more information, see [Connect offsets API reference](https://docs.confluent.io/cloud/current/ccloud/offsets-connect-v-1/).
- Use either an [Oracle System Change Number (SCN)](https://docs.oracle.com/en/database/oracle/oracle-database/19/cncpt/transactions.html#GUID-A2615547-94D2-4346-B156-64C534C5E9E4) or
  [Logical Change Record (LCR) position](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/general-xstream-concepts.html#GUID-077AAE24-54CF-4102-9969-9540608FFB65).

### Get the current offset

To get the current offset, make a `GET` request that specifies the environment, Kafka
cluster, and connector name.

```bash
GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets
Host: https://api.confluent.cloud
```

**Response:**

Successful calls return HTTP `200` with a JSON payload that describes the offset.

The following example shows the offset once the snapshot has been completed.

```bash
{
  "id": "lcc-example",
  "name": "{connector_name}",
  "offsets": [
    {
      "partition": {
        "server": "{topic_prefix}"
      },
      "offset": {
        "scn": "4374567",
        "snapshot": "INITIAL",
        "snapshot_completed": true
      }
    }
  ],
  "metadata": {
    "observed_at": "2025-03-01T12:30:00.151015100Z"
  }
}
```

The following example shows the offset once streaming is in progress.

```bash
{
  "id": "lcc-example",
  "name": "{connector_name}",
  "offsets": [
    {
      "partition": {
        "server": "{topic_prefix}"
      },
      "offset": {
        "lcr_position": "000000000044150e0000000100000001000000000044150d000000140000000102"
      }
    }
  ],
  "metadata": {
    "observed_at": "2025-03-01T12:30:00.151015100Z"
  }
}
```

Responses include the following information:

- The position of latest offset.
- The observed time of the offset in the metadata portion of the payload. The `observed_at` time
  indicates a snapshot in time for when the API retrieved the offset. A running connector is always updating
  its offsets. Use `observed_at` to get a sense for the gap between real time and the time at which the request
  was made. By default, offsets are observed every minute. Calling `GET` repeatedly will fetch more recently
  observed offsets.
- Information about the connector.

### Update the offset

To update the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector
name. Include a JSON payload that specifies new offset and a patch type.

The following example shows how to update the offset using an SCN.

```bash
POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request
Host: https://api.confluent.cloud

 {
   "type": "PATCH",
   "offsets": [
     {
       "partition": {
         "server": "{topic_prefix}"
       },
       "offset": {
         "scn": "4374567",
         "snapshot": "INITIAL",
         "snapshot_completed": true
       }
     }
   ]
 }
```

The following example shows how to update the offset using an LCR position.

```bash
POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request
Host: https://api.confluent.cloud

 {
   "type": "PATCH",
   "offsets": [
     {
       "partition": {
         "server": "{topic_prefix}"
       },
       "offset": {
         "lcr_position": "0000000000432bd400000001000000010000000000432bd3000000140000000102"
       }
     }
   ]
 }
```

Considerations:

- You can only make one offset change at a time for a given connector.
- This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information,
  see **Get the status of an offset request**.
- For source connectors, the connector attempts to read from the position defined by the requested offsets.

**Response:**

Successful calls return HTTP `202 Accepted` with a JSON payload that describes the offset.

The following example shows the response to an offset update request using an SCN.

```bash
{
  "id": "lcc-example",
  "name": "{connector_name}",
  "offsets": [
    {
      "partition": {
        "server": "{topic_prefix}"
      },
      "offset": {
        "scn": "4374567",
        "snapshot": "INITIAL",
        "snapshot_completed": true
      }
    }
  ],
  "requested_at": "2025-03-01T12:30:00.151015100Z",
  "type": "PATCH"
}
```

The following example shows the response to an offset update request using an LCR position.

```bash
{
  "id": "lcc-example",
  "name": "{connector_name}",
  "offsets": [
    {
      "partition": {
        "server": "{topic_prefix}"
      },
      "offset": {
        "lcr_position": "0000000000432bd400000001000000010000000000432bd3000000140000000102"
      }
    }
  ],
  "requested_at": "2025-03-01T12:30:00.151015100Z",
  "type": "PATCH"
}
```

Responses include the following information:

- The requested position of the offsets in the source.
- The time of the request to update the offset.
- Information about the connector.

### Delete the offset

To delete the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector
name. Include a JSON payload that specifies the delete type.

```bash
POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request
Host: https://api.confluent.cloud

 {
   "type": "DELETE"
 }
```

Considerations:

- Delete requests delete the offset for the provided partition and reset to the base state. A
  delete request is as if you created a fresh new connector.
- This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information,
  see **Get the status of an offset request**.
- Do not issue delete and patch requests at the same time.
- For source connectors, the connector attempts to read from the position defined in the base state.

**Response**:

Successful calls return HTTP `202 Accepted` with a JSON payload that describes the result.

```bash
{
  "id": "lcc-example",
  "name": "{connector_name}",
  "offsets": [],
  "requested_at": "2025-03-01T12:30:00.151015100Z",
  "type": "DELETE"
}
```

Responses include the following information:

- Empty offsets.
- The time of the request to delete the offset.
- Information about Kafka cluster and connector.
- The type of request.

### Get the status of an offset request

To get the status of a previous offset request, make a `GET` request that specifies the environment, Kafka cluster, and connector
name.

```bash
GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request/status
Host: https://api.confluent.cloud
```

Considerations:

- The status endpoint always shows the status of the most recent PATCH/DELETE operation.

**Response**:

Successful calls return HTTP `200` with a JSON payload that describes the result. The following is an example
of an applied patch.

```bash
{
  "request": {
    "id": "lcc-example",
    "name": "{connector_name}",
    "offsets": [
      {
        "partition": {
          "server": "{topic_prefix}"
        },
        "offset": {
          "scn": "2778404",
          "snapshot": "INITIAL",
          "snapshot_completed": true
        }
      }
    ],
    "requested_at": "2025-03-01T12:30:00.151015100Z",
    "type": "PATCH"
  },
  "status": {
    "phase": "APPLIED",
    "message": "The Connect framework-managed offsets for this connector have been altered successfully. However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses."
  },
  "previous_offsets": [
    {
      "partition": {
        "server": "{topic_prefix}"
      },
      "offset": {
        "lcr_position": "00000000002a8684000000010000000100000000002a8680000000010000000102"
      }
    }
  ],
  "applied_at": "2025-03-01T12:30:10.151015100Z"
}
```

Responses include the following information:

- The original request, including the time it was made.
- The status of the request: applied, pending, or failed.
- The time you issued the status request.
- The previous offsets. These are the offsets that the connector last updated
  prior to updating the offsets. Use these to try to restore the state of your connector
  if a patch update causes your connector to fail or to return a connector to its
  previous state after rolling back.

### JSON payload

You can use either an [Oracle System Change Number (SCN)](https://docs.oracle.com/en/database/oracle/oracle-database/19/cncpt/transactions.html#GUID-A2615547-94D2-4346-B156-64C534C5E9E4) or
[Logical Change Record (LCR) position](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/general-xstream-concepts.html#GUID-077AAE24-54CF-4102-9969-9540608FFB65)
to update the offsets of the Oracle XStream CDC Source connector.

The table below offers a description of the unique fields in the JSON payload for managing
offsets of the connector using an Oracle SCN.

| Field                | Definition                                                                                                                                                                                                                       | Required/Optional   |
|----------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
| `scn`                | The SCN of the change. It is set to the current SCN during the snapshot phase and<br/>obtained from the LCR position during the streaming phase.<br/><br/>To manage offsets, set this to the SCN from which to resume streaming. | Required            |
| `snapshot`           | Indicates the type of snapshot. It is set to `INITIAL` for an initial consistent snapshot.<br/><br/>To manage offsets, set this to `INITIAL`.                                                                                    | Required            |
| `snapshot_completed` | Indicates whether the snapshot has been completed.<br/><br/>To manage offsets, set this to `true`.                                                                                                                               | Required            |

The table below offers a description of the unique fields in the JSON payload for managing
offsets of the connector using an LCR position.

| Field          | Definition                                                                                                                                     | Required/Optional   |
|----------------|------------------------------------------------------------------------------------------------------------------------------------------------|---------------------|
| `lcr_position` | The position of the LCR. Set in streaming phase only.<br/><br/>To manage offsets, set this to the LCR position from which to resume streaming. | Required            |

### General considerations

- An existing outbound server can be used if the SCN or LCR position used in the offsets request
  is equal to or greater than the outbound server’s processed low position. To find the processed
  low position for an outbound server,
  see [Displaying the Processed Low Position for an Outbound Server](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/monitoring-xstream-out.html#GUID-71A5E78D-4AB8-4FA8-B1EC-C09BFC58D612).
- An existing outbound server cannot be used if the SCN or LCR position used in the offsets
  request is earlier than the outbound server’s processed low position. In this case, a new
  capture process and outbound server much be created with
  a [first SCN](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/xstream-out-concepts.html#GUID-E752A74B-DF85-4C56-A03B-8C5DF751A48D)
  and [start SCN](https://docs.oracle.com/en/database/oracle/oracle-database/19/xstrm/xstream-out-concepts.html#GUID-E752A74B-DF85-4C56-A03B-8C5DF751A48D)
  that precede the SCN or LCR position used in the offset.
  * The first SCN and start SCN must be a valid SCN and present in the redo log files available
    to the capture process.
  * The first SCN can be set to any value returned by the following query:
    * `SELECT DISTINCT FIRST_CHANGE#, NAME FROM V$ARCHIVED_LOG WHERE DICTIONARY_BEGIN = 'YES'`
    * The value returned in the `NAME` column indicates the redo log file containing the SCN
      corresponding to the first SCN. This redo log file, along with all subsequent redo log
      files, must be available to the capture process. If the query returns multiple distinct
      `FIRST_CHANGE#` values, choose the first SCN value that is most appropriate for the capture
      process you are creating.
  * The start SCN specified must be greater than or equal to the first SCN for the capture process.
- To create a new connector with offsets, set the SCN in the offset and then start the connector
  in `recovery` snapshot mode using the `snapshot.mode` configuration property. The connector will
  first capture a snapshot of the schema for the capture tables, populating the schema history
  topic, and then begin streaming from the specified SCN in the offset. Once the recovery process
  is complete and streaming begins, reset the `snapshot.mode` to either `initial` or `no_data` to
  prevent the connector from initiating a recovery upon future restarts.

  #### IMPORTANT
  The connector will fail if there have been schema changes to the captured tables after the specified SCN.
- To update the offsets of an existing connector, set the SCN or LCR position in the offset.

  #### IMPORTANT
  The connector will fail if there have been schema changes to the capture tables
  between the SCN or LCR position specified in the offsets request and the connector’s last
  processed SCN or LCR position.

### Migrate connectors

Considerations:

- The self-managed connector must be operating in streaming mode. If the self-managed connector
  is still in the process of making a snapshot, you can either create a new connector on Confluent Cloud
  which starts the snapshot process from the beginning or wait for the snapshot process to
  complete and follow the [migration guidance](../offsets.md#custom-offsets-fully-managed-migration).
- The configurations of the self-managed connector must match the configurations of the
  fully managed connector. You need to set the `snapshot.mode` configuration property to `recovery`.
  This ensures that the connector will first capture a snapshot of the schema for the capture
  tables, populating the schema history topic, and then begin streaming from the specified SCN
  in the offset.

  #### IMPORTANT
  The connector will fail if there have been schema changes to the captured tables after
  the specified SCN.

## Quick Start

Use this quick start to get up and running with the Confluent Cloud Oracle XStream CDC Source
connector. The quick start provides the basics of selecting the connector and
configuring it to obtain a snapshot of the existing data in an Oracle database
and then monitoring and recording all subsequent row-level changes.

<a id="cc-oracle-xstream-cdc-source-prereqs-main"></a>

Before configuring the connector, see [Oracle Database Prerequisites](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs) for Oracle database configuration information and post-configuration validation steps.

### Prerequisites

- Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/)
  cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud.
- The Confluent CLI installed and configured for the cluster.
  See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html).
- [Schema Registry](../../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format
  (for example, Avro, JSON_SR (JSON Schema), or Protobuf).
- For networking considerations, see [Networking and DNS](../overview.md#connect-internet-access-resources). To use a
  set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](../static-egress-ip.md#cc-static-egress-ips).

- Kafka cluster credentials. The following lists the different ways you can provide credentials.
  - Enter an existing [service account](../service-account.md#s3-cloud-service-account) resource ID.
  - Create a Confluent Cloud [service account](../service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](../service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements.
  - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.

### Using the Confluent Cloud Console

#### Step 1: Launch your Confluent Cloud cluster

To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../../get-started/index.md#cloud-create-kafka-cluster).

#### Step 2: Add a connector

In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add
connector**.

#### Step 3: Select your connector

Click the **Oracle XStream CDC Source** connector card.

![Oracle XStream CDC Source Connector Card](images/ccloud-oracle-xstream-cdc-source-icon.png)

<a id="cc-oracle-xstream-cdc-source-setup-connection"></a>

#### Step 4: Enter the connector details

#### NOTE
* Make sure you have all your [prerequisites](#cc-oracle-xstream-cdc-source-prereqs-main) completed.
* An asterisk ( \* ) designates a required entry.

At the **Add Oracle XStream CDC Source Connector** screen, complete the following:

### Kafka access

1. Select the way you want to provide **Kafka Cluster credentials**. You can
   choose one of the following options:
   - **My account**: This setting allows your connector to globally access everything
     that you have access to. With a user account, the connector uses an API key and
     secret to access the Kafka cluster. This option is not recommended for production.
   - **Service account**: This setting limits the access for your connector by using a
     [service account](../service-account.md#s3-cloud-service-account). This option is recommended for
     production.
   - **Use an existing API key**: This setting allows you to specify an API key and a
     secret pair. You can use an existing pair or create a new one. This method is not
     recommended for production environments.

   #### NOTE
   Freight clusters support only service accounts for Kafka authentication.
2. Click **Continue**.

### Authentication

1. Configure the authentication properties:

   **Authentication method**
   - **Authentication method**: Select how you want to authenticate with your database.
   - **Use secret manager**: Fetch sensitive configuration values from a secret manager.

   **Secret manager configuration**
   - **Secret manager**: Select the secret manager to use for retrieving sensitive data.
   - **Configurations from Secret manager**: Select the configurations whose values Confluent Cloud should fetch from the secret manager.
   - **Provider Integration**: Select an existing provider integration that has access to your secret manager.

   **How should we connect to your database?**
   - **Database hostname**: The IP address or hostname of the Oracle database server.
   - **Database port**: The port number used to connect to Oracle database server. Defaults
     to `1521`.
   - **Database username**: The name of the Oracle database user connecting to the Oracle database.
   - **Database password**: The password for the Oracle database user connecting to the Oracle database.
   - **Database name**: The name of the database to connect to. In a multitenant architecture,
     it refers to the container database (CDB) name.
   - **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.

     #### NOTE
     Escape special characters (for example, `$`) in the service name
     using a single backslash (`\$`).
   - **Pluggable database (PDB) name**: The name of the pluggable database (PDB) to connect to in a
     multitenant architecture. By default, this is not set, indicating that the tables to
     capture reside in the CDB root. If this is set, you must specify the CDB name in the
     `Database name` field.
   - **XStream outbound server name**: The name of the XStream outbound server to connect to.
   - **TLS mode**: Specify whether to use Transport Layer Security (TLS) to connect to the
     Oracle database.
     * `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.file` 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.file` configuration property.
   - **Client wallet file**: An 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 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.

     #### NOTE
     When using the CLI, the wallet file must be provided as a base64-encoded string.
     - Encode the `cwallet.sso` file to base64.
     - Prefix the encoded content with `data:text/plain;base64,`.
     - Use the entire string as the value for this configuration property. For example,
       `"database.wallet.file": "data:text/plain;base64,/u3+7QAAAAIAAAACAAAAAQAGY2xpZ...omitted...=="`
   - **Total number of Oracle processors to license**: The number of Oracle processor
     licenses required for the source database server or cluster. To determine this out, multiply
     the total number of processor cores by a core processor licensing factor, as mentioned in the
     Oracle Processor Core Factor Table.
   - **Downstream database hostname**: The IP address or hostname of the downstream Oracle database server. When
     specified, this enables downstream capture mode, where the snapshot and
     metadata are read from the source database and streaming occurs from the
     downstream database.
   - **Downstream database port**: The port number of the downstream Oracle database server. Required when
     downstream capture is enabled.
   - **Downstream database name**: The name of the downstream database to connect to. In a multitenant
     container database, this is the name of the container database (CDB).
     Required when downstream capture is enabled.
   - **Downstream database service name**: The name of the downstream database service to connect to. In a multitenant
     container database, this is the service used to connect to the container
     database (CDB). For Oracle Real Application Clusters (RAC), use the service
     created by Oracle XStream. Required when downstream capture is enabled.
   - **Downstream TLS mode**: Specifies whether to use Transport Layer Security (TLS) to connect to the
     downstream Oracle database. Select one of the following options:
     - `disable` (default): Does not use a TLS connection.
     - `one-way`: Uses a TLS-encrypted connection and 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 `downstream.database.wallet.location`
       configuration property.
     - `two-way`: Uses a TLS-encrypted connection and 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 `downstream.database.wallet.location` configuration
       property.
   - **Downstream client wallet file**: Specifies the Oracle client wallet (`cwallet.sso`) that holds the
     certificates used for TLS connections between the connector and the
     downstream database server. This must be a single sign-on (SSO) auto-login
     wallet. 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.
2. Click **Continue**.

### Configuration

**Output messages**

- **Output Kafka record key format**: Sets the output Kafka record key
  format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you
  need to have Confluent Cloud Schema Registry configured if using a schema-based message format
  like AVRO, JSON_SR, and PROTOBUF. Defaults to AVRO.
- **Output Kafka record value format**: Sets the output Kafka record value
  format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you
  need to have Confluent Cloud Schema Registry configured if using a schema-based message format
  like AVRO, JSON_SR, and PROTOBUF.

**How should we name your topic(s)?**

- **Topic prefix**: The topic prefix provides a namespace for the Oracle database server
  or cluster used by the connector to capture changes. It must be unique and can include only
  alphanumeric characters, hyphens, dots, and underscores. This prefix is added to all Kafka
  topic names receiving events from this connector.

  #### WARNING
  Do not change the value of this property. If changed, upon restart, the connector will
  start emitting events to new topics based on the revised value and not to the original topics,
  and it won’t be able to recover its database schema history topic.

**Connector configuration**

- **Table include list**: A comma-separated list of regular expressions that match fully-qualified
  table identifiers for the tables whose changes you want to capture. The connector will only
  capture changes from tables that match these expressions. 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.

  #### NOTE
  If you use this property, do not use the `table.exclude.list` property.
- **Table exclude list**: A comma-separated list of regular expressions that
  match fully-qualified table identifiers for the tables whose changes you do not want to capture.
  The connector will only capture 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.

  #### NOTE
  If you use this property, do not use `table.include.list` property.
- **Snapshot mode**: The criteria for running a snapshot upon startup of the connector. Select
  one of the following snapshot options:
  * **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.

  #### WARNING
  Do not use this mode to perform a snapshot if schema changes were committed to the
  database after the last connector shutdown.

**Data encryption**

- Enable **Client-Side Field Level Encryption**
  for data encryption. Specify a **Service Account** to
  access the Schema Registry and associated encryption rules or keys with that schema. For more
  information on CSFLE or CSPE setup,
  see [Manage encryption for connectors](../csfle.md#connect-csfle).

### **Show advanced configurations**

- **Schema context**: Select a schema context to use for this connector, if using
  a schema-based data format. This property defaults to the **Default** context,
  which configures the connector to use the default schema set up for Schema Registry in your
  Confluent Cloud environment. A schema context allows you to use separate schemas (like
  schema sub-registries) tied to topics in different Kafka clusters that share the
  same Schema Registry environment. For example, if you select a non-default context, a
  **Source** connector uses only that schema context to register a schema and a
  **Sink** connector uses only that schema context to read from. For more
  information about setting up a schema context, see [What are schema contexts and when should you use them?](../../sr/faqs-cc.md#faq-schema-contexts).

**Additional Configs**

- **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`.
- **Errors Tolerance**: Use this property to configure the connector’s error handling behavior.

  #### WARNING
  Use this property with caution for sink connectors, as it can lead to data loss. If you set this property to `all`, the connector does not fail on errant records, but logs them (and sends to DLQ for sink connectors) and continues processing. If you set this property to `none`, the connector task fails on errant records.
- **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals:
  `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and
  `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value.
- **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
- **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters.
- **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
- **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration.
- **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.

**Auto-restart policy**

- **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its
  task in the event of user-actionable errors. Defaults to `true`, enabling the connector to
  automatically restart in case of user-actionable errors. Set this property to `false` to
  disable auto-restart for failed connectors. If disabled, you must manually restart the connector.

**Connector configuration**

- **Skip unparseable DDL statement**: 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.

  #### WARNING
  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.
- **Max retries on snapshot database errors**: 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.
- **Emit tombstone on delete**: Controls whether a delete event is followed by a tombstone
  event. The following values are possible:
  * **true (default)**: 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.

  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.
- **Skipped operations**: A comma-separated list of the operation types to skip
  during streaming. You can configure the connector to skip the following types
  of operations:
  * `c` (create/insert)
  * `u` (update)
  * `d` (delete)
  * `t` (truncate)

  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. The following values
  are possible:
  * **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. The following values
  are possible:
  * **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.
- **Heartbeat interval (ms)**: Controls the frequency of heartbeat messages sent by the
  connector to a heartbeat topic.
  * 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.
- **Database server’s operating system (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.
- **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.

  #### 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.

  #### NOTE
  Primary key columns are always included in an event’s key, even if you use this property to explicitly exclude its value.
- **Unavailable value placeholder**: Specifies the constant that the connector uses to indicate that the
  original value is unavailable and not provided by the database. Defaults to `__cflt_unavailable_value`.
- **Oversized large object (LOB) 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.
- **Oversized large object (LOB) handling mode**: Defines how the connector handles LOB column values that exceed the
  size threshold specified using the `lob.oversize.threshold` configuration. Select one of the following options:
  * **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. Defaults to `__cflt_skipped_value`.
- **Signal table name**: 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.

  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.

  When connecting to a Pluggable Database (PDB), use the name of the PDB as the database name
  component of the fully-qualified name.
- **Signal enabled channels**: Specifies which signaling channels are enabled for the connector.
  Supported values are:
  * **source** (default): Signals are read from a signaling table in the source database.
  * **kafka**: Signals are consumed from a Kafka topic.

  You can enable multiple channels by providing a comma-separated list. If not set, the
  connector enables only the `source` channel by default.
- **Signal topic name**: 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.
- **Outbound server attach position mode**: Specifies the position the connector passes when attaching to the XStream outbound server.
  Select one of the following options:
  - `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.
  - `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.

  #### NOTE
  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 (as it would in `connector_managed` mode). The
  reset has no effect, the connector resumes from the outbound server’s
  processed low position.

  Change the default value only after consulting Confluent support.
- **Snapshot select statement overrides data map**: An optional JSON object that maps fully-qualified table identifiers to custom SELECT statements that determine which rows to include in the initial snapshot. Each identifier is of the form `schemaName.tableName`. 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. Tables configured with a custom SELECT statement are read as of the statement’s own execution SCN rather than the shared snapshot SCN, and are therefore not captured with the same point-in-time consistency as other tables.
  The case of object names in the fully-qualified table identifier must match exactly 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 and must be enclosed in double quotes. Any double quotes inside the key or value must be escaped with a backslash.

**How should we handle data types?**

- **Decimal handling mode**: Specifies how the connector should handle NUMBER, DECIMAL and
  NUMERIC columns. You can set one of the following options:
  * **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.
  You can set one of the following options:
  * **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. You can set one of the following options:
  * **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.

**Transforms**

- **Single Message Transformations**: To add a new SMT, see [Add transforms](../single-message-transforms.md#cc-single-message-transforms-ui).
  For more information about unsupported SMTs, see
  [Unsupported transformations](../single-message-transforms.md#cc-single-message-transforms-unsupported-transforms).

**Processing position**

- **Set offsets**: Click **Set offsets** to define a specific offset for
  this connector to begin procession data from. For more information
  on managing offsets, see [Manage offsets](../offsets.md#connect-custom-offsets).

For all property values and definitions, see [Configuration Properties](#cc-oracle-xstream-cdc-source-config-properties).

- Click **Continue**.

### Sizing

Based on the number of topic partitions you select, you will be provided
with a recommended number of tasks.

1. To change the number of tasks, enter the desired number of tasks for the connector to
   use in the **Maximum number of tasks** field.

   #### NOTE
   The connector always operates with a single task.
2. Click **Continue**.

### Review and Launch

> 1. Verify the connection details by previewing the running configuration.
> 2. After you’ve validated that the properties are configured to your
>    satisfaction, click **Launch**.

>    The status for the connector should go from **Provisioning** to
>    **Running**.

If the connector is not running, see [Oracle Database Prerequisites](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs) and
review the Oracle database configuration information and post-configuration validation steps.

#### Step 5: Check the Kafka topic

After the connector is running, verify that records are populating your Kafka topic.

For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section.

### Using the Confluent CLI

Complete the following steps to set up and run the connector using the Confluent CLI.

#### NOTE
* Make sure you have all your [prerequisites](#cc-oracle-xstream-cdc-source-prereqs-main) completed.
* The example commands use Confluent CLI version 2. For more information
  see, [Confluent CLI v2](https://docs.confluent.io/confluent-cli/current/migrate.html#cli-migrate).

#### Step 1: List the available connectors

Enter the following command to list available connectors:

```none
confluent connect plugin list
```

#### Step 2: List the connector configuration properties

Enter the following command to show the connector configuration properties:

```none
confluent connect plugin describe <connector-plugin-name>
```

The command output shows the required and optional configuration properties.

For example:

```none
confluent connect plugin describe OracleXStreamSource
```

Example output:

```none
The following are required configs:
connector.class : OracleXStreamSource
database.dbname
database.hostname
database.service.name
database.out.server.name
database.user
database.password
name
output.data.value.format
tasks.max
topic.prefix
kafka.api.key : ["kafka.api.key" is required when "kafka.auth.mode==KAFKA_API_KEY"]
kafka.api.secret : ["kafka.api.secret" is required when "kafka.auth.mode==KAFKA_API_KEY"]
```

#### Step 3: Create the connector configuration file

Create a JSON file that contains the connector configuration properties to deploy an instance
of the connector.

The connector setup below performs the following:

- Connects to the `ORCLPDB1` pluggable database located at `db.example.com`
  on port `1521`.
- Initiates a snapshot of the `employees` table in the `sample`
  schema within the `ORCLPDB1` pluggable database.
- After snapshot completion, the connector listens for changes made to the `employees` table
  through the `XOUT` outbound server.
- Streams the changes to the Kafka topic, `cflt.SAMPLE.EMPLOYEES`.

```none
{
   "name": "oracle-connector",
   "config": {
     "connector.class": "io.confluent.connect.oracle.xstream.cdc.OracleXStreamSourceConnector",
     "tasks.max" : "1",
     "database.hostname": "db.example.com",
     "database.port": "1521",
     "database.user": "C##CFLTUSER",
     "database.password": "secret",
     "database.dbname": "ORCLCDB",
     "database.service.name": "ORCLCDB",
     "database.pdb.name": "ORCLPDB1",
     "database.out.server.name": "XOUT",
     "table.include.list": "SAMPLE.EMPLOYEES",
     "topic.prefix": "cflt",
     "kafka.auth.mode": "KAFKA_API_KEY",
     "kafka.api.key": "****************",
     "kafka.api.secret": "**************************************************"
   }
 }
```

Note the following property definitions:

* `"name"`: Sets a name for your new connector.
* `"connector.class"`: Identifies the connector plugin name.

* `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above).  To use a [service account](../service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=<service-account-resource-ID>`. To list the available service account resource IDs, use the following command:
  ```bash
  confluent iam service-account list
  ```

  For example:
  ```bash
  confluent iam service-account list

     Id     | Resource ID |       Name        |    Description
  +---------+-------------+-------------------+-------------------
     123456 | sa-l1r23m   | sa-1              | Service account 1
     789101 | sa-l4d56p   | sa-2              | Service account 2
  ```

* `"database.service.name"`: The name of the database service
  to connect to. If the service name contains a special character (for
  example, `$`), escape it with a double backslash (`\\$`) in the JSON configuration
  file. For example, a service name of `ORCLCDB$1` is written as
  `"database.service.name": "ORCLCDB\\$1"`.

#### NOTE
To enable CSFLE or CSPE for data encryption, specify the following properties:

* `csfle.enabled`: Flag to indicate whether the connector honors CSFLE or CSPE rules.
* `sr.service.account.id`: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema.

For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](../csfle.md#connect-csfle).

**SMTs**: To add a new SMT using the Confluent CLI, see [Add transforms](../single-message-transforms.md#cc-single-message-transforms-cli).
For more information about the Debezium SMT **ExtractNewRecordState**, see [Debezium transformations](../single-message-transforms.md#cc-single-message-transforms-debezium-unwrap).

See [Configuration Properties](#cc-oracle-xstream-cdc-source-config-properties) for all properties and definitions.

#### Step 4: Load the properties file and create the connector

Enter the following command to load the configuration and start the connector:

```none
confluent connect cluster create --config-file <file-name>.json
```

For example:

```none
confluent connect cluster create --config-file oracle-xstream-cdc-source.json
```

Example output:

```none
Created connector OracleXStreamSource_0 lcc-ix4dl
```

#### Step 5: Check the connector status

Enter the following command to check the connector status:

```none
confluent connect cluster list
```

Example output:

```none
ID         |            Name       | Status  |  Type
+----------+-----------------------+---------+--------+
lcc-ix4dl  | OracleXStreamSource_0 | RUNNING | source
```

#### Step 6: Check the Kafka topic.

After the connector is running, verify that messages are populating your Kafka topic.

If the connector is not running, see [Oracle Database Prerequisites for Oracle XStream CDC Source Connector for Confluent Cloud](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs) and review the
Oracle database configuration information and post-configuration validation steps.

For more information and examples to use with the Confluent Cloud API for Connect,
see the [Confluent Cloud API for Connect Usage Examples](../connect-api-section.md#ccloud-connect-api) section.

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

<a id="oracle-cdc-xstream-source-snapshot-by-table-partitions"></a>

## Configuration Properties

Use the following configuration properties with the fully managed Oracle XStream CDC Source connector.

### How should we connect to your data?

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

### Kafka Cluster credentials

`kafka.auth.mode`
: Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
  <br/>
  * Type: string
  * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
  * Importance: high

`kafka.api.key`
: Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
  <br/>
  * Type: password
  * Importance: high

`kafka.service.account.id`
: The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
  <br/>
  * Type: string
  * Importance: high

`kafka.api.secret`
: Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
  <br/>
  * Type: password
  * Importance: high

### Schema Config

`schema.context.name`
: Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
  <br/>
  * Type: string
  * Default: default
  * Importance: medium

### Authentication method

`authentication.method`
: Select how you want to authenticate with your database.
  <br/>
  * Type: string
  * Default: Password
  * Importance: high

`secret.manager.enabled`
: Fetch sensitive configuration values from a secret manager.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: high

### Secret manager configuration

`secret.manager`
: Select the secret manager to use for retrieving sensitive data.
  <br/>
  * Type: string
  * Importance: high

`secret.manager.managed.configs`
: Select the configurations to fetch their values from the secret manager.
  <br/>
  * Type: list
  * Importance: high

`secret.manager.provider.integration.id`
: Select an existing provider integration that has access to your secret manager.
  <br/>
  * Type: string
  * Importance: high

### How should we connect to your database?

`database.hostname`
: IP address or hostname of the Oracle database server.
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^[^\?=%&\(\)\\$]*$`
  * 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
  * Valid Values: Must match the regex `^[^\?=%&\(\)]*$`
  * 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 container database, this is the name of the container database (CDB).
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^[a-zA-Z][a-zA-Z0-9$#_]*$`
  * 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 container database (CDB). For Oracle Real Application Clusters (RAC), use the service created by Oracle XStream. Escape special characters (for example, `$`) in the service name using a single backslash (`\$`).
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^[a-zA-Z](?:[a-zA-Z0-9#._]|\\\$)*$`
  * 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
  * Valid Values: Must match the regex `^([a-zA-Z][a-zA-Z0-9$#_]*)*$`
  * Importance: high

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

`database.tls.mode`
: Specifies whether to use Transport Layer Security (TLS) to connect to the Oracle database. Select one of the following options:
  <br/>
  disable (default): Does not use a TLS connection.
  <br/>
  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.
  <br/>
  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.
  <br/>
  * Type: string
  * Default: disable
  * Valid Values: disable, one-way, two-way
  * Importance: medium

`database.wallet.file`
: Specifies the Oracle client wallet (cwallet.sso) that holds 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 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.
  <br/>
  * Type: password
  * Default: [hidden]
  * Importance: medium

`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. Confluent shares this value with Oracle for audit purposes. Enter the exact number of processor licenses for your Oracle database.
  <br/>
  * Type: int
  * Valid Values: [1,…,4096]
  * Importance: medium

`downstream.database.hostname`
: IP address or hostname of the downstream Oracle database server. When specified, enables downstream capture mode where the snapshot and metadata are read from the source database, and streaming occurs from the downstream database.
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^[^\?=%&\(\)\\$]*$`
  * Importance: medium

`downstream.database.port`
: Port number of the downstream Oracle database server. Required when downstream capture is enabled.
  <br/>
  * Type: int
  * Default: 1521
  * Valid Values: [1,…,65535]
  * Importance: medium

`downstream.database.dbname`
: Name of the downstream database to connect to. In a multitenant container database, this is the name of the container database (CDB). Required when downstream capture is enabled.
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^([a-zA-Z][a-zA-Z0-9$#_]*)*$`
  * Importance: medium

`downstream.database.service.name`
: Name of the downstream database service to which to connect. In a multitenant container database, this is the service used to connect to the container database (CDB). For Oracle Real Application Clusters (RAC), use the service created by Oracle XStream. Required when downstream capture is enabled.
  <br/>
  * Type: string
  * Valid Values: Must match the regex `^([a-zA-Z](?:[a-zA-Z0-9#._]|\\\$)*)*$`
  * Importance: medium

`downstream.database.tls.mode`
: Specifies whether to use Transport Layer Security (TLS) to connect to the downstream Oracle database. Select one of the following options:
  <br/>
  disable (default): Does not use a TLS connection.
  <br/>
  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 downstream.database.wallet.location configuration property.
  <br/>
  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 downstream.database.wallet.location configuration property.
  <br/>
  * Type: string
  * Default: disable
  * Valid Values: disable, one-way, two-way
  * Importance: medium

`downstream.database.wallet.file`
: Specifies the Oracle client wallet (cwallet.sso) that holds the certificates used for TLS connections between the connector and the downstream database server. This must be a single sign-on (SSO) auto-login wallet. 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.
  <br/>
  * Type: password
  * Default: [hidden]
  * Importance: medium

### Output messages

`output.key.format`
: Sets the output Kafka record key format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured when using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
  <br/>
  * Type: string
  * Default: AVRO
  * Valid Values: AVRO, JSON_SR, PROTOBUF
  * Importance: high

`output.data.format`
: Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, or PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured when using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.
  <br/>
  * Type: string
  * Default: AVRO
  * Valid Values: AVRO, JSON_SR, PROTOBUF
  * Importance: high

### How should we name your topic(s)?

`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/>
  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.
  <br/>
  * Type: string
  * Importance: high

### 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: high

`snapshot.mode`
: The criteria for running a snapshot upon startup of the connector. Select one of the following snapshot options:
  <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.
  <br/>
  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.
  <br/>
  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. WARNING: Do not use this mode to perform a snapshot if schema changes were committed to the database after the last connector shutdown.
  <br/>
  * Type: string
  * Default: initial
  * Valid Values: initial, no_data, recovery
  * Importance: medium

`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

`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

`tombstones.on.delete`
: Controls whether a delete event is followed by a tombstone event. The following values are possible:
  <br/>
  true: For each delete operation, the connector emits a delete event and a subsequent tombstone event.
  <br/>
  false: For each delete operation, the connector emits only a delete event.
  <br/>
  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

`skipped.operations`
: A comma-separated list of operations to skip during streaming. You can configure the connector to skip the following types of operations: c (inserts/create), u (updates), d (deletes), t (truncates), and none to indicate nothing is skipped. The default value is t, ensuring that only truncate operations are skipped.
  <br/>
  * Type: string
  * Default: t
  * Importance: low

`schema.name.adjustment.mode`
: Specifies how schema names should be adjusted for compatibility with the message converter used by the connector. The following values are possible:
  <br/>
  none (the default) does not apply any adjustment.
  <br/>
  avro replaces the characters that cannot be used in the Avro type name with underscore.
  <br/>
  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.
  <br/>
  * Type: string
  * Default: none
  * Valid Values: avro, avro_unicode, none
  * Importance: low

`field.name.adjustment.mode`
: Specifies how field names should be adjusted for compatibility with the message converter used by the connector. The following values are possible:
  <br/>
  none (the default) does not apply any adjustment.
  <br/>
  avro replaces the characters that cannot be used in the Avro type name with underscore.
  <br/>
  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.
  <br/>
  * Type: string
  * Default: none
  * Valid Values: avro, avro_unicode, none
  * 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

`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

`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 include this property in the configuration, do not set the column.exclude.list property. 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.
  <br/>
  * Type: string
  * Importance: medium

`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 include this property in the configuration, do not set the column.include.list property. Note: Primary key columns are always included in an event’s key, even if you use this property to explicitly exclude its value.
  <br/>
  * Type: string
  * Importance: medium

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

`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
  * Valid Values: [-1,…]
  * Importance: low

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

`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 in the format: databaseName.schemaName.tableName.
  <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 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. Supported values are:
  <br/>
  source: Signals are read from a signaling table in the source database.
  <br/>
  kafka: Signals are consumed from a Kafka topic.
  <br/>
  You can enable multiple channels by providing a comma-separated list. If not set, only the source channel is enabled by default.
  <br/>
  * Type: list
  * Default: source
  * Importance: medium

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

`snapshot.select.statement.overrides.data.map`
: An optional JSON object that maps fully-qualified table identifiers to custom SELECT statements that determine which rows to include in the initial snapshot. Each identifier is of the form `schemaName.tableName`. 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. Tables configured with a custom SELECT statement are read as of the statement’s own execution SCN rather than the shared snapshot SCN, and are therefore not captured with the same point-in-time consistency as other tables.
  <br/>
  The case of object names in the fully-qualified table identifier must match exactly 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 and must be enclosed in double quotes. Any double quotes inside the key or value must be escaped with a backslash.
  <br/>
  * Type: password
  * Importance: medium

`outbound.server.attach.position.mode`
: Specifies the position the connector passes when attaching to the XStream outbound server. 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.
  <br/>
  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 (as it would in connector_managed mode). The reset has no effect, the connector resumes from the outbound server’s processed low position.
  <br/>
  Note: Change the default value only after consulting Confluent support.
  <br/>
  * Type: string
  * Default: connector_managed
  * Importance: low

### How should we handle data types?

`decimal.handling.mode`
: Specifies how the connector should handle NUMBER, DECIMAL and NUMERIC columns. You can set one of the following options:
  <br/>
  precise (the 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.
  <br/>
  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.
  <br/>
  double: Represents values using Java’s double. Using double values is easier, but can result in a loss of precision.
  <br/>
  * Type: string
  * Default: precise
  * Valid Values: double, precise, string
  * Importance: medium

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

`time.precision.mode`
: Specifies how the connector should handle time, date, and timestamp columns. You can set one of the following options:
  <br/>
  adaptive (the default): Bases the precision of time, date, and timestamp values on the database column’s precision.
  <br/>
  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.
  <br/>
  * Type: string
  * Default: adaptive
  * Valid Values: adaptive, connect
  * Importance: medium

### Number of tasks for this connector

`tasks.max`
: Specifies the maximum number of tasks for the connector. Since this connector supports only a single task, the maximum is capped at 1.
  <br/>
  * Type: int
  * Default: 1
  * Valid Values: [1,…,1]
  * Importance: high

### Additional Configs

`header.converter`
: The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
  <br/>
  * Type: string
  * Importance: low

`producer.override.compression.type`
: The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
  <br/>
  * Type: string
  * Importance: low

`producer.override.linger.ms`
: The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
  <br/>
  * Type: long
  * Valid Values: [100,…,1000]
  * Importance: low

`value.converter.allow.optional.map.keys`
: Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.auto.register.schemas`
: Specify if the Serializer should attempt to register the Schema.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.connect.meta.data`
: Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.avro.schema.support`
: Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.enhanced.protobuf.schema.support`
: Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.flatten.unions`
: Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.index.for.unions`
: Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.generate.struct.for.nulls`
: Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.int.for.enums`
: Whether to represent enums as integers. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.latest.compatibility.strict`
: Verify latest subject version is backward compatible when use.latest.version is true.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.object.additional.properties`
: Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.nullables`
: Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.optional.for.proto2`
: Whether proto2 optionals are supported. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.latest.version`
: Use latest version of schema in subject for serialization when auto.register.schemas is false.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.use.optional.for.nonrequired`
: Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.wrapper.for.nullables`
: Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`value.converter.wrapper.for.raw.primitives`
: Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
  <br/>
  * Type: boolean
  * Importance: low

`errors.tolerance`
: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
  <br/>
  * Type: string
  * Default: none
  * Importance: low

`key.converter.key.schema.id.serializer`
: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
  * Importance: low

`key.converter.key.subject.name.strategy`
: How to construct the subject name for key schema registration.
  <br/>
  * Type: string
  * Default: TopicNameStrategy
  * Importance: low

`value.converter.decimal.format`
: Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
  <br/>
  BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
  <br/>
  NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
  <br/>
  * Type: string
  * Default: BASE64
  * Importance: low

`value.converter.flatten.singleton.unions`
: Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`value.converter.reference.subject.name.strategy`
: Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
  <br/>
  * Type: string
  * Default: DefaultReferenceSubjectNameStrategy
  * Importance: low

`value.converter.value.schema.id.serializer`
: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
  <br/>
  * Type: string
  * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer
  * Importance: low

`value.converter.value.subject.name.strategy`
: Determines how to construct the subject name under which the value schema is registered with Schema Registry.
  <br/>
  * Type: string
  * Default: TopicNameStrategy
  * Importance: low

### Auto-restart policy

`auto.restart.on.user.error`
: Enable connector to automatically restart on user-actionable errors.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: medium

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

## Connect to an Oracle Real Application Cluster (RAC) Database

Confluent recommends configuring the following properties to ensure that the connector
will be able to connect and attach to the specific RAC instance
running the XStream components:

- Configure the `database.hostname` property to the Oracle RAC database SCAN address.

  #### NOTE
  If a SCAN address is unavailable, configure the `database.hostname` property to the
  hostname of the instance where the XStream components are running. You will need to
  manually reconfigure the connector whenever the instance running
  the XStream components changes.
- Configure the `database.service.name` property to the auto-created Oracle XStream service.

  You can find the service name using the following query:
  ```sql
  SELECT NETWORK_NAME FROM DBA_QUEUES q, DBA_XSTREAM_OUTBOUND ob
    WHERE q.OWNER = ob.QUEUE_OWNER AND q.NAME = ob.QUEUE_NAME AND SERVER_NAME = '<outbound_server>';
  ```

  Replace the placeholder `<outbound_server>` with the actual outbound server name.

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

## Supported Data Types

The connector creates change events for database changes. Each change event mirrors the
table’s schema, with a field for every column value. The data type of each table column
determines how the connector represents the column values in the corresponding change event fields.

For certain data types, such as numeric data types, you can customize how the connector maps
them by modifying the default configuration settings. This allows more control over handling
various data types, ensuring that the change events reflect the desired format and meet specific
requirements.

### Character data types

The following table describes how the connector maps character types.

| Oracle data type   | Connect type   |
|--------------------|----------------|
| CHAR               | STRING         |
| VARCHAR / VARCHAR2 | STRING         |
| NCHAR              | STRING         |
| NVARCHAR           | STRING         |

In all cases, the connector ensures that character data is converted to a string type
in Kafka Connect when creating change events.

### Large object (LOB) types

You can adjust how the connector maps binary LOB data types by changing the `binary.handling.mode` configuration property.

The following table describes how the connector maps binary and character LOB types.

| Oracle data type   | Connect type   | Notes                                                                                                                                                                                                                                                                    |
|--------------------|----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| BLOB               | BYTES          | Based on `binary.handling.mode`:<br/><br/>* `bytes`: Represents as byte array (default)<br/>* `base64`: Represents as base64-encoded string<br/>* `base64-url-safe`: Represents as base64-url-safe-encoded string<br/>* `hex`: Represents as hex-encoded (base16) string |
| CLOB               | STRING         |                                                                                                                                                                                                                                                                          |
| NCLOB              | STRING         |                                                                                                                                                                                                                                                                          |

### Numeric data types

You can adjust how the connector maps numeric data types by changing the `decimal.handling.mode`
configuration property.

The table below shows the mapping of numeric types when `decimal.handling.mode` is set to `precise`.

| Oracle data type                                                        | Connect type                         | Notes                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
|-------------------------------------------------------------------------|--------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| NUMBER(P, S <= 0)                                                       | INT8 / INT16 / INT32 / INT64 / BYTES | Based on the precision and scale, the connector selects a matching Kafka Connect integer type:<br/><br/>* If the precision minus the scale (P - S) is less than 3, it uses INT8.<br/>* If P - S is less than 5, it uses INT16.<br/>* If P - S is less than 10, it uses INT32.<br/>* If P - S is less than 19, it uses INT64.<br/>* If P - S is 19 or greater, it uses BYTES (org.apache.kafka.connect.data.Decimal).<br/><br/>NUMBER columns with a scale of `0` represent integer numbers. A negative scale indicates<br/>rounding in Oracle, for example, a scale of `-2` causes rounding to hundreds. |
| NUMBER(P, S > 0)                                                        | BYTES                                | org.apache.kafka.connect.data.Decimal                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
| NUMBER(P, [, \* ])                                                      | STRUCT                               | io.debezium.data.VariableScaleDecimal<br/><br/>Contains a structure with two fields: scale (of type INT32) that contains the scale<br/>of the transferred value, and value (of type BYTES) containing the original value in<br/>an unscaled form.                                                                                                                                                                                                                                                                                                                                                        |
| SMALLINT, INT, INTEGER                                                  | BYTES                                | org.apache.kafka.connect.data.Decimal<br/><br/>Oracle maps SMALLINT, INT and INTEGER to NUMBER(38,0). As a result, these types can<br/>hold values that exceed the maximum range of any of the INT types.                                                                                                                                                                                                                                                                                                                                                                                                |
| NUMERIC, DECIMAL                                                        | INT8 / INT16 / INT32 / INT64 / BYTES | Handles in the same way as the NUMBER data type (note that scale defaults to `0` for NUMERIC).                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
| FLOAT[(P)]<br/><br/>Maps to FLOAT(126) when P not mentioned             | STRUCT                               | io.debezium.data.VariableScaleDecimal<br/><br/>Contains a structure with two fields: scale (of type INT32) that contains the scale of the<br/>transferred value, and value (of type BYTES) containing the original value in an unscaled form.                                                                                                                                                                                                                                                                                                                                                            |
| REAL - Maps to FLOAT(63)<br/><br/>DOUBLE PRECISION - Maps to FLOAT(126) | STRUCT                               | io.debezium.data.VariableScaleDecimal<br/><br/>Contains a structure with two fields: scale (of type INT32) that contains the scale of<br/>the transferred value, and value (of type BYTES) containing the original value in an unscaled form.                                                                                                                                                                                                                                                                                                                                                            |
| BINARY_FLOAT                                                            | FLOAT32                              |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
| BINARY_DOUBLE                                                           | FLOAT64                              |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |

#### NOTE
When `decimal.handling.mode` is set to:

- string: The Oracle numeric data types are mapped to the Kafka Connect `STRING` type.
- double: The Oracle numeric data types are mapped to the Kafka Connect `FLOAT64` type.

### Temporal data types

You can adjust how the connector maps some of the temporal data types by changing the `time.precision.mode`
configuration property.

The table below shows the mapping of temporal types:

| Oracle data type                  | Connect type   | Notes                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|-----------------------------------|----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| DATE                              | INT64          | Based on `time.precision.mode`:<br/><br/>* `adaptive`: io.debezium.time.Timestamp<br/>* `connect`: org.apache.kafka.connect.data.Timestamp                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
| TIMESTAMP[(P)]                    | INT64          | Based on `time.precision.mode`:<br/><br/>**adaptive**:<br/><br/>If precision <= 3: io.debezium.time.Timestamp<br/><br/>* Represents the number of milliseconds since the UNIX epoch, without timezone information.<br/><br/>Else if precision <= 6: io.debezium.time.MicroTimestamp<br/><br/>* Represents the number of microseconds since the UNIX epoch, without timezone information.<br/><br/>Else: io.debezium.time.NanoTimestamp<br/><br/>* Represents the number of nanoseconds since the UNIX epoch, without timezone information.<br/><br/>**connect**:<br/><br/>org.apache.kafka.connect.data.Timestamp<br/><br/>Represents the number of milliseconds since the UNIX epoch, without timezone information. |
| TIMESTAMP WITH TIMEZONE           | STRING         | io.debezium.time.ZonedTimestamp<br/><br/>A string representation of a timestamp with timezone information.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
| TIMESTAMP WITH LOCAL TIMEZONE     | STRING         | io.debezium.time.ZonedTimestamp<br/><br/>A string representation of a timestamp in UTC.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |
| INTERVAL YEAR[(P)] TO MONTH       | STRING         | io.debezium.time.Interval<br/><br/>A string representation of the interval value in the ISO 8601 duration format: P<years>Y<months>M<days>DT<hours>H<minutes>M<seconds>S.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
| INTERVAL DAY[(P)] TO SECOND[(FP)] | STRING         | io.debezium.time.Interval<br/><br/>A string representation of the interval value in the ISO 8601 duration format: P<years>Y<months>M<days>DT<hours>H<minutes>M<seconds>S.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |

#### NOTE
When `time.precision.mode` is set to `connect`, there could be a loss of precision if the
fractional second precision of a column exceeds `3`, because Oracle supports a higher level
of precision than the logical types in Kafka Connect.

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

## Security

### Native Network Encryption

Oracle database provides native network encryption and integrity to ensure data is secure during
transit, without the need for setting up Transport Layer Security (TLS).

Encryption and integrity are managed based on a combination of client-side and server-side
encryption settings, which can be configured using parameters in the `sqlnet.ora` configuration
file. For more information on configuring Oracle Advanced Security for network encryption and
integrity, see [Support for Network Encryption and Integrity](https://docs.oracle.com/en/database/oracle/oracle-database/19/jjdbc/client-side-security.html#GUID-42C16CD9-62D3-4141-BB38-015627A49079) section in the Oracle
Database JDBC Developer’s Guide.

The connector uses the Oracle JDBC OCI driver to communicate with the Oracle database. You can
use the database’s native network encryption and data integrity to securely transmit data
between the connector and the Oracle database. For more information on relevant configuration
settings, see [Table 9-2 OCI Driver Client Parameters for Encryption and Integrity](https://docs.oracle.com/en/database/oracle/oracle-database/19/jjdbc/client-side-security.html#GUID-DA1DB155-D904-4D6D-9376-C499BF6EFC61)
in the Oracle Database JDBC Developer’s Guide.

The following configurations are set on the connector:

- `SQLNET.ALLOW_WEAK_CRYPTO` parameter is set to FALSE to ensure that the connector uses
  strong algorithms when communicating with the Oracle database.
- `SQLNET.ENCRYPTION_CLIENT` and `SQLNET.CRYPTO_CHECKSUM_CLIENT` parameters are set to
  ACCEPTED (the default value).

To enable network encryption and integrity, configure the `SQLNET.ENCRYPTION_SERVER` and
`SQLNET.CRYPTO_CHECKSUM_SERVER` parameters on the server to either REQUESTED or REQUIRED.
Additionally, specify strong encryption and crypto-checksum algorithms by setting the
`SQLNET.ENCRYPTION_TYPES_SERVER` and `SQLNET.CRYPTO_CHECKSUM_TYPES_SERVER` parameters.
For more information, see [Improving Native Network Encryption Security](https://docs.oracle.com/en/database/oracle/oracle-database/19/dbseg/configuring-network-data-encryption-and-integrity.html#GUID-19089E9B-6FDE-44C2-8D79-DCFAF33E91BC)
section in the Oracle Database Security Guide.

### Transport Layer Security (TLS)

You can configure Transport Layer Security (TLS) to secure connections between the
client (connector) and the Oracle database. The connector supports both one-way TLS
(with and without client wallets) and two-way TLS (mutual TLS).

- **One-way TLS**: In one-way TLS, the database server presents a certificate to authenticate itself to the
  connector. The connector needs access to the trusted Certificate Authority (CA)
  root certificate that signed the server’s certificate to verify it. This trusted CA root
  certificate must be available either in the default certificate store on Confluent Cloud or
  within a client wallet provided to the connector. Note that the default certificate store
  includes only root certificates from well-known CAs.
- **Two-way TLS** (Mutual TLS): In two-way TLS, both the connector and the database server
  present certificates to authenticate each other. The connector needs access to the
  trusted CA root certificate that signed the server’s certificate, and the server must
  have the trusted CA root certificate that signed the client certificate. You must store
  the client certificate and the trusted CA root certificate (used to
  sign the server’s certificate) within a client wallet and provide this wallet to the connector.

For more information on configuring TLS, see
[Configuring Transport Layer Security Authentication](https://docs.oracle.com/en/database/oracle/oracle-database/19/dbseg/configuring-secure-sockets-layer-authentication.html)
chapter of the Oracle Database Security Guide.

You can enable TLS connections between the connector and the Oracle database server by
using the `database.tls.mode` configuration property:

- Set `database.tls.mode` to `one-way` to enable TLS encryption and server authentication.
- Set `database.tls.mode` to `two-way` to enable TLS encryption with both client and server authentication.

To use a client wallet, provide a Single Sign-On (SSO) auto-login wallet file (named `cwallet.sso`)
using the `database.wallet.file` configuration property. This wallet must contain the certificates
used for TLS connections between the connector and the database server.

#### NOTE
When `database.tls.mode` is set to `one-way` or `two-way`, ensure that the port specified in
`database.port` corresponds to the listener on the server that supports TLS connections.

When using downstream capture, the connector establishes two independent
TLS connections: one to the source database and one to the downstream
database. Configure TLS for the downstream connection separately using
`downstream.database.tls.mode` and `downstream.database.wallet.file`.
Any combination of `disable`, `one-way`, or `two-way` is valid for
each connection independently. For more information, see
[Transport Layer Security (TLS)](downstream-capture.md#xstream-downstream-capture-tls).

<a id="cc-oracle-xstream-cdc-source-tde"></a>

### Transparent Data Encryption (TDE)

Transparent Data Encryption (TDE) enables you to encrypt sensitive data stored in tables and
tablespaces. The data is transparently decrypted for authorized users or applications during access.

The connector supports capturing changes from databases encrypted with Oracle TDE. Oracle XStream
performs decryption, and the connector receives decrypted data from the XStream outbound sever.

In multitenant environments, only united mode is supported. In this mode, a single shared keystore
is configured for the CDB root and any associated united mode PDBs.

#### NOTE
For Oracle XStream to encrypt columns transparently, the encryption master key must be stored
in the keystore on the local database, and the keystore must be open.

When using downstream capture, the downstream database must share the source’s TDE keystore by
using a Network File System (NFS) or a manual copy of the wallet file. Sharing the keystore
allows the XStream capture process to decrypt redo data from encrypted source tablespaces
and columns. You must update the downstream database whenever the keystore changes on
the source database.

## Suggested Reading

Blog post: [Oracle XStream CDC Connector](https://www.confluent.io/blog/Oracle-XStream-CDC-Connector/)

## Next Steps

For an example that shows fully managed Confluent Cloud connectors in action with
Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html).
This example also shows how to use Confluent CLI to manage your resources in
Confluent Cloud.

[![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)