<a id="cp-oracle-cdc-qs-quickstart"></a>

# Quick Start for the Oracle CDC Source Connector for Confluent Platform

Use the Kafka Connect Oracle CDC Source connector to capture changes to rows
in a database, and then represent those changes as change event records in
Apache Kafka® topics.

In this quick start, you will:

- [Verify prerequisites](#cp-oracle-cdc-qs-prereq)
- [Launch the connector](#cp-oracle-cdc-qs-step1)
- [Verify the connector processes records](#cp-oracle-cdc-qs-step2)
- [Produce change events](#cp-oracle-cdc-qs-step3)
- [Consume from topics](#cp-oracle-cdc-qs-step4)
- [Delete the connector and topics](#cp-oracle-cdc-qs-step5)
- [Troubleshoot the connector](#cp-oracle-cdc-qs-step6)

<a id="cp-oracle-cdc-qs-prereq"></a>

## Prerequisites

Before you proceed with the Oracle CDC connector quick start, ensure you do the
following:

- [Install Confluent Platform](https://docs.confluent.io/platform/current/platform-quickstart.html)
- [Install the Oracle CDC Source connector plugin for Confluent Platform](overview.md#connect-oracle-cdc-source-installation)
- [Complete the Oracle database prerequisites](prereqs-validation.md#connect-oracle-cdc-source-prereqs)
- [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html)

<a id="cp-oracle-cdc-qs-step1"></a>

### Step 1: Launch the connector

Use the steps in this section to launch the Oracle CDC Source connector on Confluent Platform
and produce data to Kafka topics. You can launch the connector using either of
the following methods:

### Confluent CLI

1. Load the connector by passing a `.json` configuration file.
   ```none
   confluent local services connect connector load <name-of-connector> --config <path-to-config-file>
   ```

   Example of a configuration file:
   ```none
   {
     "name": "cdc-source-oracle-pdb",
     "config": {
       "connector.class": "io.confluent.connect.oracle.cdc.OracleCdcSourceConnector",
       "tasks.max": "2",
       "key.converter": "io.confluent.connect.avro.AvroConverter",
       "value.converter": "io.confluent.connect.avro.AvroConverter",
       "oracle.server": "localhost",
       "oracle.port": "1521",
       "oracle.sid": "ORCLCDB",
       "oracle.pdb.name": "ORCLPDB1",
       "oracle.username": "C##MYUSER",
       "oracle.password": "mypassword",
       "redo.log.consumer.bootstrap.servers": "localhost:9092",
       "table.inclusion.regex": "ORCLPDB1[.]C##MYUSER[.].*",
       "value.converter.schema.registry.url": "http://localhost:8081",
       "key.converter.schema.registry.url": "http://localhost:8081"
     }
   }
   ```

   #### NOTE
   Note that these are the minimum configurations required to run the
   connector. For detailed configurations, see [configuration
   properties](configuration-properties.md#connect-oracle-cdc-source-config).
2. Verify the connector configurations.
   ```none
   confluent local services connect connector config <name-of-connector>
   ```

   The output should be similar to:
   ```none
   Current configuration of cdc-oracle-source-pdb:
   {
      "connector.class": "io.confluent.connect.oracle.cdc.OracleCdcSourceConnector",
      "oracle.password": "mypassword",
      "oracle.server": "localhost",
      "oracle.sid": "ORCLCDB",
      "oracle.pdb.name": "ORCLPDB1",
      "redo.log.consumer.bootstrap.servers": "localhost:9092",
      "tasks.max": "2",
      "oracle.port": "1521",
      "oracle.username": "C##MYUSER",
      "value.converter.schema.registry.url": "http://localhost:8081",
      "table.inclusion.regex": "ORCLPDB1[.]C##MYUSER[.].*",
      "name": "cdc-source-oracle-pdb",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://localhost:8081"
   }
   ```

### Control Center (Legacy)

1. Navigate to Control Center (Legacy) at [http://localhost:9021](http://localhost:9021). It may take a minute or two for Control Center (Legacy)
   to start and load.
2. Click the **controlcenter.cluster** tile.
3. In the navigation menu, click **Connect**.
4. Click the `connect-default` cluster in the **Connect clusters** list.
5. Click **Add connector** to start creating a connector.
6. Select the **OracleCDCSourceConnector** tile

   #### NOTE
   You can also click on **Upload connector config file** and select a `.json`
   file from your local machine containing the connector configuration. For example:
   ```none
   {
     "name": "cdc-source-oracle-pdb",
     "config": {
       "connector.class": "io.confluent.connect.oracle.cdc.OracleCdcSourceConnector",
       "tasks.max": "2",
       "key.converter": "io.confluent.connect.avro.AvroConverter",
       "value.converter": "io.confluent.connect.avro.AvroConverter",
       "oracle.server": "localhost",
       "oracle.port": "1521",
       "oracle.sid": "ORCLCDB",
       "oracle.pdb.name": "ORCLPDB1",
       "oracle.username": "C##MYUSER",
       "oracle.password": "mypassword",
       "redo.log.consumer.bootstrap.servers": "localhost:9092",
       "table.inclusion.regex": "ORCLPDB1[.]C##MYUSER[.].*",
       "value.converter.schema.registry.url": "http://localhost:8081",
       "key.converter.schema.registry.url": "http://localhost:8081"
     }
   }
   ```
7. In the **Name** field, enter `cdc-oracle-source-pdb` as the name of the connector.
8. Enter the following configuration values in the following sections:

   **Name** section: `cdc-source-oracle-pdb`

   **Common** section:
   - **Tasks Max:** `2`
   - **Key converter class:** `io.confluent.connect.avro.AvroConverter`
   - **Value converter class:** `io.confluent.connect.avro.AvroConverter`

   **Oracle Connection** section:
   - **Oracle Server:** `localhost`
   - **Oracle Port:** `1521`
   - **Oracle PDB Name** `ORCLPDB1`
   - **SID:** `ORCLCDB`
   - **Oracle Username:** `C##MYUSER`
   - **Oracle Password:** `mypassword`

   **Oracle Redo Logs** section:
   - **Redo Log Consumer Bootstrap Servers:** `localhost:9092`

   **Oracle Tables** section:
   - **Table inclusion Regex:** `ORCLPDB1[.]C##MYUSER[.].*`

   **Additional Properties** section:
   - **key.converter.schema.registry.url:** `http://localhost:8081`
   - **value.converter.schema.registry.url:** `http://localhost:8081`

   #### NOTE
   Note that these are the minimum configurations required to run the
   connector. For detailed configurations, see [configuration
   properties](configuration-properties.md#connect-oracle-cdc-source-config).
9. Click **Next** to review the connector configuration. If satisfied with
   the settings, click **Launch**, else click **Back** and modify the
   settings.

<a id="cp-oracle-cdc-qs-step2"></a>

### Step 2: Verify the connector processes records

After [launching the connector](#cp-oracle-cdc-qs-step1), the redo log
topic and table topics are auto created. Use the following steps to verify the
connector is processing records.

Note that the connector can run in snapshot and/or change data capture (CDC)
mode depending on the value set for the `start.from` [configuration property](configuration-properties.md#connect-oracle-cdc-source-config). This quick start does snapshot mode first
and then switches to CDC mode (default).

The connector will first read the accessible tables–tables that are included in
`table.inclusion.regex` and not in `table.exclusion.regex`—and find the
tables for which the snapshot is not complete. It will initiate the snapshot for
all those tables.

#### NOTE
If the validations fail for any reason, such as invalid configurations or
unmet [database prerequisites](prereqs-validation.md#connect-oracle-cdc-source-prereqs), the
connector will go in a failed state and the following steps would not occur.

1. Verify the connector creates a topic for each table, based on the
   `table.topic.name.template` configuration property. The connect logs should
   resemble the following:
   ```none
   Determining status of snapshots for tables ORCLPDB1.C##MYUSER.CUSTOMERS
   Found 0 of 1 snapshots are complete: [table 'ORCLPDB1.C##MYUSER.CUSTOMERS' at SCN=2640682]
   Beginning 1 snapshots using 1 of 4 snapshot threads
   Reading 2301 rows from table 'ORCLPDB1.C##MYUSER.CUSTOMERS' at SCN=2640682
   Completed snapshot of 2301 rows from table 'ORCLPDB1.C##MYUSER.CUSTOMERS' at SCN=2642470 in 0:00:02.130
   Completed snapshots for all assigned tables after 0:00:02.707.
   ```
2. Once the snapshot is complete, verify the connector proceeds to capture
   change events. You should see connect logs that resemble the following:
   ```none
   Proceeding to capture change events for ORCLPDB1.C##MYUSER.CUSTOMERS since FULL Supplemental Logging mode is set
   ```

<a id="cp-oracle-cdc-qs-step3"></a>

### Step 3: Produce change events

Use the steps in this section to produce change events.

1. To produce a change event, execute a DML statement on the database connected to
   the connector and commit the command as shown in the following example:
   ```none
   SQL> insert into CUSTOMERS (id, first_name, last_name, email, gender, club_status, comments) values (10001, 'Rica', 'Blaisdell', 'rblaisdell0@rambler.ru', 'Female', 'bronze', 'Universal optimal hierarchy');

   1 row created.

   SQL> commit;

   Commit complete.
   ```

   #### IMPORTANT
   Ensure you commit the data for a change event to be processed and produced
   to the topic. Transactions that have been rolled back will not be produced
   to the topic.

   In case no new records are being generated in the table specific topics, see
   [Why do my table-specific topics show no new records?](https://docs.confluent.io/kafka-connectors/oracle-cdc/current/troubleshooting.html#why-do-my-table-specific-topics-show-no-new-records).
2. Verify the redo log topic has been created after the first DML post
   connector creation and snapshot in accordance with the
   `redo.log.topic.name` property.

<a id="cp-oracle-cdc-qs-step4"></a>

### Step 4: Consume from topics

Use this section to consume data from Kafka topics. You can use the Confluent Platform GUI or
the Confluent CLI.

### Confluent CLI

1. Start consuming from the topic. To consume from the beginning of the
   topic append the flag `--from-beginning`:
   ```none
   confluent local services kafka consume <topic-name> --from-beginning
   ```

### Confluent Platform GUI

1. Navigate to Control Center (Legacy) at [http://localhost:9021](http://localhost:9021). It may take a minute or two for Control Center (Legacy)
   to start and load.
2. Click the **controlcenter.cluster** tile.
3. In the navigation menu, click **Topics**.
4. Click on the topic from which to consume `<topic-name>`.
5. Navigate to the **Messages** tab. All the new messages produced are
   visible here. For viewing all messages from beginning for a partition,
   enter `0/Partition:0` as the offset.

<a id="cp-oracle-cdc-qs-step5"></a>

### Step 5: Delete the Oracle CDC connector and topics

Use this section to delete an Oracle CDC Source connector and its associated
Kafka topics. You can use the Confluent Platform GUI or the Confluent CLI.

### Confluent CLI

1. Unload the connector to delete it.
   ```none
   confluent local services connect connector unload <connector-name>
   ```

### Confluent Platform GUI

1. Navigate to Control Center (Legacy) at [http://localhost:9021](http://localhost:9021). It may take a minute or two for Control Center (Legacy)
   to start and load.
2. Click the **controlcenter.cluster** tile.
3. In the navigation menu, click **Connect**.
4. Click the `connect-default` cluster in the **Connect clusters** list.
5. Click the connector that needs to be deleted.
6. Click **Delete**.
7. Enter the name of the connector to be deleted for confirmation.
8. Click **Confirm** to delete the connector.

<a id="cp-oracle-cdc-qs-step6"></a>

### Step 6: Troubleshoot the Oracle CDC connector

For connect related logging and debugging, run the following command and append
the `--follow` flag to log additional output until the command is interrupted:

```none
confluent local services connect log -f
```

For more troubleshooting tips, see [Troubleshooting](https://docs.confluent.io/kafka-connectors/oracle-cdc/current/troubleshooting.html).