<a id="connect-oracle-xstream-cdc-signal-actions"></a>

# Signals and Actions for Oracle XStream CDC Source Connector for Confluent Platform

The connector offers a signaling mechanism to trigger specific actions, such as
performing an ad-hoc snapshot of a table.

You can send signals through one or more channels, such as a database table or a Kafka
topic. To configure which channels are enabled, use the `signal.enabled.channels` configuration property.

## Channels

You must explicitly configure signaling for each connector you want to use it with. The
required configuration varies depending on the signaling channel you use.

### Source signaling channel

To set up a source signaling channel, create a signaling table with a specific schema in the
source database. To send a signal, insert a record into this table. The connector enables this channel by default.

Each connector instance must have its own unique signaling table.

#### Create a signaling table

In the source database, create a signaling table to send signals to the connector.
The table must follow the structure below, with columns in the exact order shown:

| Column   | Type     | Notes                                                                            |
|----------|----------|----------------------------------------------------------------------------------|
| id       | VARCHAR2 | A unique identifier for the signal instance, typically a UUID string.            |
| type     | VARCHAR2 | The type of signal being sent.                                                   |
| data     | VARCHAR2 | JSON-formatted parameters that provide additional details for the signal action. |

The following example creates a signaling table named `cflt_signals` that follows the required structure:

```sql
CREATE TABLE cflt_signals (
  id VARCHAR(42) PRIMARY KEY,
  type VARCHAR(32) NOT NULL,
  data VARCHAR(2048)
);
```

Additionally, ensure the following:

1. Enable supplemental logging for this table. For detailed instruction, see [Configure supplemental
   logging](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs-supplemental-logging).
2. The database user configured on the connector has the necessary privileges on this table.
   For detailed instructions, see [Configure database users](prereqs-validation.md#connect-oracle-xstream-cdc-source-prereqs-user-privileges).

#### Add the signaling table to Oracle XStream rules

To ensure that signals added to the signaling table are captured and sent to the connector
during streaming, you must add the signaling table to the rule set of both
the capture process and the outbound server.

For more information on adding tables to the rule sets of the capture process and outbound
server, see [Configure the table capture set](examples.md#connect-oracle-xstream-cdc-examples-configure-table).

#### Add the signaling table to connector configuration

Set the `signal.data.collection` configuration property to the fully-qualified name of
the signaling table, using the format: `<databaseName>.<schemaName>.<tableName>`.

#### NOTE
The case of the 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.

For more information about the `signal.data.collection` property,
see [Connector configuration properties](configuration-properties.md#connect-oracle-xstream-cdc-source-connector-config).

### Kafka signaling channel

To configure a Kafka signaling channel, add `kafka` to the `signal.enabled.channels`
configuration property and specify the name of the Kafka topic that will receive
signals using the `signal.kafka.topic` property.

Each connector instance must have its own unique signaling topic. The topic must have
a single partition to ensure signal ordering is preserved. Once you enable this channel,
the connector creates a Kafka consumer to read signals from the configured topic.

#### Kafka signal requirements

Each Kafka message used as a signal must meet the following requirements:

- The message key must match the value of the `topic.prefix` configuration property.
- The message value must be a JSON object containing the `type` and `data` fields.

Use the `StringConverter` for both the key and value when producing records to the signaling topic.

#### Configure the signal topic consumer

You can configure the Kafka consumer that the connector uses to read from the signal topic by
using pass-through configuration properties with the prefix `signal.consumer`. The connector
removes this prefix before passing the properties to the consumer. For example, to set the
signal consumer’s `security.protocol` property to `SSL`, use `signal.consumer.security.protocol=SSL`.

## Actions

You can use signaling to instruct the connector to perform specific actions. This section
describes the supported actions.

<a id="xstream-signal-ad-hoc-blocking-snapshot"></a>

### Ad-hoc blocking snapshot

An ad-hoc blocking snapshot captures a snapshot of specific tables after the
connector has already started streaming changes. This action is useful when
you update the connector configuration to include new tables or need to backfill
missing data into Kafka topics.

To trigger an ad-hoc blocking snapshot, send a signal with `type` set to `execute-snapshot`
and `data.type` set to `blocking`. When the connector receives this signal, it temporarily
pauses streaming, performs the snapshot using the same process as the initial snapshot,
and then resumes streaming once the snapshot is complete.

#### NOTE
A short delay may occur between when the signal is sent and when streaming pauses for the snapshot to begin.
As a result, after the snapshot completes, the connector might emit some change events
that duplicate those captured in the snapshot.

Within the signal, you can specify a subset of tables from the source database to include
in the snapshot. For each table, you can also define additional conditions to control which
records are captured. All tables specified in the signal must be part of the connector’s table inclusion list.

The following table describes the structure of the `data` element in an ad-hoc blocking snapshot signal:

| Field                 | Default   | Notes                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
|-----------------------|-----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| type                  | n/a       | Specifies the type of snapshot to run. Currently, only `blocking` snapshots are supported.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
| data-collections      | n/a       | An array of regular expressions matching the fully-qualified names of tables to snapshot,<br/>using the format: `<databaseName>.<schemaName>.<tableName>`.<br/><br/>#### NOTE<br/>The case of the object names must exactly match how they are stored in the Oracle database.<br/>Unquoted identifiers are case-insensitive and are treated as uppercase by default, whereas<br/>quoted identifiers are case-sensitive.                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
| additional-conditions | n/a       | (Optional) An array that defines additional conditions the connector uses to determine<br/>which records to include in the snapshot. Each additional condition is an object that<br/>specifies the filtering criteria.<br/><br/>You can set the following parameters for each additional condition:<br/><br/>- *data-collection*: The fully-qualified name of the table the filter applies to.<br/>  The case must exactly match how they are stored in the Oracle database, following the same<br/>  case sensitivity rules as above.<br/>- *filter*: A SELECT statement that determines the subset of table rows to include<br/>  in the snapshot. The case of schema and table names must exactly match how they<br/>  are stored in the Oracle database, following the same case sensitivity rules as above.<br/>  Quoted identifiers must be enclosed in double quotes. |

#### NOTE
XMLType columns are supported only when they are explicitly serialized to
a CLOB. If a `filter` statement queries an `XMLType` column, query the
column using
`XMLSERIALIZE(DOCUMENT column_name AS CLOB NO INDENT) AS column_name`,
replacing `column_name` with the name of the `XMLType` column.

When an ad-hoc snapshot is initiated for a table that already has an associated Kafka topic, the
connector appends the snapshot data to that topic. If the topic was deleted or does not exist,
the connector can automatically create a new one if automatic topic creation is enabled.

Change events generated during an ad-hoc blocking snapshot have the `snapshot` field in the
`source` block set to a value other than `false`, and the `op` field set to `r`, indicating a READ operation.

#### Using a source channel

To initiate an ad-hoc blocking snapshot using the source channel, insert a signal record
into the signaling table on the source database.

The following example shows how to send a signal to initiate an ad-hoc blocking snapshot
for the tables `ORDERS` and `ORDER_LINE` in the `OE` schema of the `ORCLPDB1` database:

```sql
INSERT INTO cflt_signals (id, type, data)
VALUES (
  '6af62216-7086-439b-b915-51825f9e49f7',
  'execute-snapshot',
  '{
      "type": "blocking",
      "data-collections": ["ORCLPDB1.OE.ORDERS", "ORCLPDB1.OE.ORDER_LINE"]
   }'
);
```

If you want the snapshot to include only a subset of data from a table, you can add
an `additional-conditions` parameter to the signal.

The following example initiates an ad-hoc blocking snapshot that captures orders from
the `ORDERS` table for a specific customer with ID 42, and order lines from the `ORDER_LINE`
table where the quantity ordered is greater than 5:

```sql
INSERT INTO cflt_signals (id, type, data)
VALUES (
  '8a474adc-caca-4cf4-85d7-2b42e056ef5c',
  'execute-snapshot',
  '{
      "type": "blocking",
      "data-collections": ["ORCLPDB1.OE.ORDERS", "ORCLPDB1.OE.ORDER_LINE"],
      "additional-conditions": [
        {
          "data-collection": "ORCLPDB1.OE.ORDERS",
          "filter": "SELECT * FROM OE.ORDERS WHERE o_c_id = 42"
        },
        {
          "data-collection": "ORCLPDB1.OR.ORDER_LINE",
          "filter": "SELECT * FROM OE.ORDER_LINE WHERE ol_quantity > 5"
        }
      ]
    }'
);
```

#### Using a Kafka channel

To initiate an ad-hoc blocking snapshot using the Kafka channel, send a signal message
to the signaling topic configured on the connector.

The key of the Kafka message must match the value of the `topic.prefix` connector configuration
property. The value of the message must be a JSON object that includes both `type` and `data` fields.

The following example shows a Kafka signal message value that initiates an ad-hoc blocking snapshot
for the tables `ORDERS` and `ORDER_LINE` in the `OE` schema of the `ORCLPDB1` database.

```json
{
  "type":"execute-snapshot",
  "data": {
    "type": "BLOCKING",
    "data-collections": ["ORCLPDB1.OE.ORDERS", "ORCLPDB1.OE.ORDER_LINE"]
   }
}
```

To snapshot only a subset of data from a table, add an `additional-conditions` parameter to
the signal. This allows you to define specific filters for each table using SQL SELECT statements.

The following example initiates an ad-hoc blocking snapshot that captures orders from the `ORDERS`
table for a specific customer with ID 42, and order lines from the `ORDER_LINE` table where the
quantity ordered is greater than 5:

```json
{
  "type":"execute-snapshot",
  "data": {
    "type": "BLOCKING",
    "data-collections": ["ORCLPDB1.OE.ORDERS", "ORCLPDB1.OE.ORDER_LINE"],
    "additional-conditions": [
     {
       "data-collection": "ORCLPDB1.OE.ORDERS",
       "filter": "SELECT * FROM OE.ORDERS WHERE o_c_id = 42"
     },
     {
       "data-collection": "ORCLPDB1.OE.ORDER_LINE",
       "filter": "SELECT * FROM OE.ORDER_LINE WHERE ol_quantity > 5"
     }
    ]
  }
}
```