Debezium Connector for Oracle

Overview

Debezium’s Oracle connector captures and records row-level changes that occur in databases on an Oracle server, including tables that are added while the connector is running. You can configure the connector to emit change events for specific subsets of schemas and tables, or to ignore, mask, or truncate values in specific columns.

For information about the Oracle Database versions that are compatible with this connector, see the Debezium release overview.

Debezium ingests change events from Oracle by using the native LogMiner database package or the XStream API. While the connector might work with a variety of Oracle versions and editions, only Oracle EE 12 and 19 have been tested.

How the Oracle connector works

To optimally configure and run a Debezium Oracle connector, it is helpful to understand how the connector performs snapshots, streams change events, determines Kafka topic names, uses metadata, and implements event buffering.

Snapshots

Typically, the redo logs on an Oracle server are configured to not retain the complete history of the database. As a result, the Debezium Oracle connector cannot retrieve the entire history of the database from the logs. To enable the connector to establish a baseline for the current state of the database, the first time that the connector starts, it performs an initial consistent snapshot of the database.

You can customize the way that the connector creates snapshots by setting the value of the snapshot.mode connector configuration property. By default, the connector’s snapshot mode is set to initial.

Default connector workflow for creating an initial snapshot

When the snapshot mode is set to the default, the connector completes the following tasks to create a snapshot:

1. Determines the tables to be captured.
2. Obtains a ROW SHARE MODE lock on each of the monitored tables to prevent structural changes from occurring during creation of the snapshot. Debezium holds the locks for only a short time.
3. Reads the current system change number (SCN) position from the server’s redo log.
4. Captures the structure of all relevant tables.
5. Releases the locks obtained in Step 2.
6. Scans all of the relevant database tables and schemas as valid at the SCN position that was read in Step 3 (SELECT * FROM …​ AS OF SCN 123), generates a READ event for each row, and then writes the event records to the table-specific Kafka topic.
7. Records the successful completion of the snapshot in the connector offsets.

After the snapshot process begins, if the process is interrupted due to connector failure, rebalancing, or other reasons, the process restarts after the connector restarts. After the connector completes the initial snapshot, it continues streaming from the position that it read in Step 3 so that it does not miss any updates. If the connector stops again for any reason, after it restarts, it resumes streaming changes from where it previously left off.

Table 1. Settings for snapshot.mode connector configuration property

For more information, see snapshot.mode in the table of connector configuration properties.

Ad hoc snapshots

By default, a connector runs an initial snapshot operation only after it starts for the first time. Following this initial snapshot, under normal circumstances, the connector does not repeat the snapshot process. Any future change event data that the connector captures comes in through the streaming process only.

However, in some situations the data that the connector obtained during the initial snapshot might become stale, lost, or incomplete. To provide a mechanism for recapturing table data, Debezium includes an option to perform ad hoc snapshots. The following changes in a database might be cause for performing an ad hoc snapshot:

• The connector configuration is modified to capture a different set of tables.
• Kafka topics are deleted and must be rebuilt.
• Data corruption occurs due to a configuration error or some other problem.

You can re-run a snapshot for a table for which you previously captured a snapshot by initiating a so-called ad-hoc snapshot. Ad hoc snapshots require the use of signaling tables. You initiate an ad hoc snapshot by sending a signal request to the Debezium signaling table.

When you initiate an ad hoc snapshot of an existing table, the connector appends content to the topic that already exists for the table. If a previously existing topic was removed, Debezium can create a topic automatically if automatic topic creation is enabled.

Ad hoc snapshot signals specify the tables to include in the snapshot. The snapshot can capture the entire contents of the database, or capture only a subset of the tables in the database.

You specify the tables to capture by sending an execute-snapshot message to the signaling table. Set the type of the execute-snapshot signal to incremental, and provide the names of the tables to include in the snapshot, as described in the following table:

Table 2. Example of an ad hoc execute-snapshot signal record

| Field | Default | Value |

| --- | --- | --- | | type | incremental | Specifies the type of snapshot that you want to run. Setting the type is optional. Currently, you can request only incremental snapshots. | | data-collections | N/A | An array that contains the fully-qualified names of the table to be snapshotted. The format of the names is the same as for the signal.data.collection configuration option. |

Triggering an ad hoc snapshot

You initiate an ad hoc snapshot by adding an entry with the execute-snapshot signal type to the signaling table. After the connector processes the message, it begins the snapshot operation. The snapshot process reads the first and last primary key values and uses those values as the start and end point for each table. Based on the number of entries in the table, and the configured chunk size, Debezium divides the table into chunks, and proceeds to snapshot each chunk, in succession, one at a time.

Currently, the execute-snapshot action type triggers incremental snapshots only. For more information, see Incremental snapshots.

Incremental snapshots

To provide flexibility in managing snapshots, Debezium includes a supplementary snapshot mechanism, known as incremental snapshotting. Incremental snapshots rely on the Debezium mechanism for sending signals to a Debezium connector. Incremental snapshots are based on the DDD-3 design document.

In an incremental snapshot, instead of capturing the full state of a database all at once, as in an initial snapshot, Debezium captures each table in phases, in a series of configurable chunks. You can specify the tables that you want the snapshot to capture and the size of each chunk. The chunk size determines the number of rows that the snapshot collects during each fetch operation on the database. The default chunk size for incremental snapshots is 1 KB.

As an incremental snapshot proceeds, Debezium uses watermarks to track its progress, maintaining a record of each table row that it captures. This phased approach to capturing data provides the following advantages over the standard initial snapshot process:

• You can run incremental snapshots in parallel with streamed data capture, instead of postponing streaming until the snapshot completes. The connector continues to capture near real-time events from the change log throughout the snapshot process, and neither operation blocks the other.
• If the progress of an incremental snapshot is interrupted, you can resume it without losing any data. After the process resumes, the snapshot begins at the point where it stopped, rather than recapturing the table from the beginning.
• You can run an incremental snapshot on demand at any time, and repeat the process as needed to adapt to database updates. For example, you might re-run a snapshot after you modify the connector configuration to add a table to its table.include.list property.

Incremental snapshot process

When you run an incremental snapshot, Debezium sorts each table by primary key and then splits the table into chunks based on the configured chunk size. Working chunk by chunk, it then captures each table row in a chunk. For each row that it captures, the snapshot emits a READ event. That event represents the value of the row when the snapshot for the chunk began.

As a snapshot proceeds, it’s likely that other processes continue to access the database, potentially modifying table records. To reflect such changes, INSERT, UPDATE, or DELETE operations are committed to the transaction log as per usual. Similarly, the ongoing Debezium streaming process continues to detect these change events and emits corresponding change event records to Kafka.

How Debezium resolves collisions among records with the same primary key

In some cases, the UPDATE or DELETE events that the streaming process emits are received out of sequence. That is, the streaming process might emit an event that modifies a table row before the snapshot captures the chunk that contains the READ event for that row. When the snapshot eventually emits the corresponding READ event for the row, its value is already superseded. To ensure that incremental snapshot events that arrive out of sequence are processed in the correct logical order, Debezium employs a buffering scheme for resolving collisions. Only after collisions between the snapshot events and the streamed events are resolved does Debezium emit an event record to Kafka.

Snapshot window

To assist in resolving collisions between late-arriving READ events and streamed events that modify the same table row, Debezium employs a so-called snapshot window. The snapshot windows demarcates the interval during which an incremental snapshot captures data for a specified table chunk. Before the snapshot window for a chunk opens, Debezium follows its usual behavior and emits events from the transaction log directly downstream to the target Kafka topic. But from the moment that the snapshot for a particular chunk opens, until it closes, Debezium performs a de-duplication step to resolve collisions between events that have the same primary key..

For each data collection, the Debezium emits two types of events, and stores the records for them both in a single destination Kafka topic. The snapshot records that it captures directly from a table are emitted as READ operations. Meanwhile, as users continue to update records in the data collection, and the transaction log is updated to reflect each commit, Debezium emits UPDATE or DELETE operations for each change.

As the snapshot window opens, and Debezium begins processing a snapshot chunk, it delivers snapshot records to a memory buffer. During the snapshot windows, the primary keys of the READ events in the buffer are compared to the primary keys of the incoming streamed events. If no match is found, the streamed event record is sent directly to Kafka. If Debezium detects a match, it discards the buffered READ event, and writes the streamed record to the destination topic, because the streamed event logically supersede the static snapshot event. After the snapshot window for the chunk closes, the buffer contains only READ events for which no related transaction log events exist. Debezium emits these remaining READ events to the table’s Kafka topic.

The connector repeats the process for each snapshot chunk.

Triggering an incremental snapshot

Currently, the only way to initiate an incremental snapshot is to send an ad hoc snapshot signal to the signaling table on the source database. You submit signals to the table as SQL INSERT queries. After Debezium detects the change in the signaling table, it reads the signal, and runs the requested snapshot operation.

The query that you submit specifies the tables to include in the snapshot, and, optionally, specifies the kind of snapshot operation. Currently, the only valid option for snapshots operations is the default value, incremental.

To specify the tables to include in the snapshot, provide a data-collections array that lists the tables, for example, {"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}

The data-collections array for an incremental snapshot signal has no default value. If the data-collections array is empty, Debezium detects that no action is required and does not perform a snapshot.

Prerequisites

• Signaling is enabled.

• A signaling data collection exists on the source database and the connector is configured to capture it.

• The signaling data collection is specified in the signal.data.collection property.

Procedure

1. Send a SQL query to add the ad hoc incremental snapshot request to the signaling table:

1. INSERT INTO _<signalTable>_ (id, type, data) VALUES (_'<id>'_, _'<snapshotType>'_, '{"data-collections": ["_<tableName>_","_<tableName>_"],"type":"_<snapshotType>_"}');

For example,

1. INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.table1", "schema2.table2"],"type":"incremental"}');

The values of the id,type, and data parameters in the command correspond to the fields of the signaling table.

The following table describes the these parameters:

Table 3. Descriptions of fields in a SQL command for sending an incremental snapshot signal to the signaling table

  | Value | Description |

| --- | --- | | myschema.debezium_signal | Specifies the fully-qualified name of the signaling table on the source database | | ad-hoc-1 | The id parameter specifies an arbitrary string that is assigned as the id identifier for the signal request. Use this string to identify logging messages to entries in the signaling table. Debezium does not use this string. Rather, during the snapshot, Debezium generates its own id string as a watermarking signal. | | execute-snapshot | Specifies type parameter specifies the operation that the signal is intended to trigger. | | data-collections | A required component of the data field of a signal that specifies an array of table names to include in the snapshot. The array lists tables by their fully-qualified names, using the same format as you use to specify the name of the connector’s signaling table in the signal.data.collection configuration property. | | incremental | An optional type component of the data field of a signal that specifies the kind of snapshot operation to run. Currently, the only valid option is the default value, incremental. Specifying a type value in the SQL query that you submit to the signaling table is optional. If you do not specify a value, the connector runs an incremental snapshot. |

The following example, shows the JSON for an incremental snapshot event that is captured by a connector.

Example: Incremental snapshot event message

1. {
2. "before":null,
3. "after": {
4. "pk":"1",
5. "value":"New data"
6. },
7. "source": {
8. ...
9. "snapshot":"incremental" (1)
10. },
11. "op":"r", (2)
12. "ts_ms":"1620393591654",
13. "transaction":null
14. }

Topic names

By default, the Oracle connector writes change events for all INSERT, UPDATE, and DELETE operations that occur in a table to a single Apache Kafka topic that is specific to that table. The connector uses the following convention to name change event topics:

serverName.schemaName.tableName

The following list provides definitions for the components of the default name:

serverName

The logical name of the server as specified by the database.server.name connector configuration property.

schemaName

The name of the schema in which the operation occurred.

tableName

The name of the table in which the operation occurred.

For example, if fulfillment is the server name, inventory is the schema name, and the database contains tables with the names orders, customers, and products, the Debezium Oracle connector emits events to the following Kafka topics, one for each table in the database:

1. fulfillment.inventory.orders
2. fulfillment.inventory.customers
3. fulfillment.inventory.products

The connector applies similar naming conventions to label its internal database history topics, schema change topics, and transaction metadata topics.

If the default topic name do not meet your requirements, you can configure custom topic names. To configure custom topic names, you specify regular expressions in the logical topic routing SMT. For more information about using the logical topic routing SMT to customize topic naming, see Topic routing.

Schema change topic

You can configure a Debezium Oracle connector to produce schema change events that describe structural changes that are applied to captured tables in the database. The connector writes schema change events to a Kafka topic named *<serverName>*, where *serverName* is the logical server name that is specified in the database.server.name configuration property.

Debezium emits a new message to this topic whenever it streams data from a new table.

Messages that the connector sends to the schema change topic contain a payload, and, optionally, also contain the schema of the change event message. The payload of a schema change event message includes the following elements:

ddl

Provides the SQL CREATE, ALTER, or DROP statement that results in the schema change.

databaseName

The name of the database to which the statements are applied. The value of databaseName serves as the message key.

tableChanges

A structured representation of the entire table schema after the schema change. The tableChanges field contains an array that includes entries for each column of the table. Because the structured representation presents data in JSON or Avro format, consumers can easily read messages without first processing them through a DDL parser.

Debezium emits a new message to this topic whenever it streams data from a new table, or when the structure of the table is altered.

Example: Message emitted to the Oracle connector schema change topic

The following example shows a typical schema change message in JSON format. The message contains a logical representation of the table schema.

1. {
2. "schema": {
3. ...
4. },
5. "payload": {
6. "source": {
7. "version": "1.9.6.Final",
8. "connector": "oracle",
9. "name": "server1",
10. "ts_ms": 1588252618953,
11. "snapshot": "true",
12. "db": "ORCLPDB1",
13. "schema": "DEBEZIUM",
14. "table": "CUSTOMERS",
15. "txId" : null,
16. "scn" : "1513734",
17. "commit_scn": "1513754",
18. "lcr_position" : null,
19. "rs_id": "001234.00012345.0124",
20. "ssn": 1,
21. "redo_thread": 1
22. },
23. "databaseName": "ORCLPDB1", (1)
24. "schemaName": "DEBEZIUM", //
25. "ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n   (    \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n    \"FIRST_NAME\" VARCHAR2(255), \n    \"LAST_NAME" VARCHAR2(255), \n    \"EMAIL\" VARCHAR2(255), \n     PRIMARY KEY (\"ID\") ENABLE, \n     SUPPLEMENTAL LOG DATA (ALL) COLUMNS\n   ) SEGMENT CREATION IMMEDIATE \n  PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 \n NOCOMPRESS LOGGING\n  STORAGE(INITIAL 65536 NEXT 1048576 MINEXTENTS 1 MAXEXTENTS 2147483645\n  PCTINCREASE 0 FREELISTS 1 FREELIST GROUPS 1\n  BUFFER_POOL DEFAULT FLASH_CACHE DEFAULT CELL_FLASH_CACHE DEFAULT)\n  TABLESPACE \"USERS\" ", (2)
26. "tableChanges": [ (3)
27. {
28. "type": "CREATE", (4)
29. "id": "\"ORCLPDB1\".\"DEBEZIUM\".\"CUSTOMERS\"", (5)
30. "table": { (6)
31. "defaultCharsetName": null,
32. "primaryKeyColumnNames": [ (7)
33. "ID"
34. ],
35. "columns": [ (8)
36. {
37. "name": "ID",
38. "jdbcType": 2,
39. "nativeType": null,
40. "typeName": "NUMBER",
41. "typeExpression": "NUMBER",
42. "charsetName": null,
43. "length": 9,
44. "scale": 0,
45. "position": 1,
46. "optional": false,
47. "autoIncremented": false,
48. "generated": false
49. },
50. {
51. "name": "FIRST_NAME",
52. "jdbcType": 12,
53. "nativeType": null,
54. "typeName": "VARCHAR2",
55. "typeExpression": "VARCHAR2",
56. "charsetName": null,
57. "length": 255,
58. "scale": null,
59. "position": 2,
60. "optional": false,
61. "autoIncremented": false,
62. "generated": false
63. },
64. {
65. "name": "LAST_NAME",
66. "jdbcType": 12,
67. "nativeType": null,
68. "typeName": "VARCHAR2",
69. "typeExpression": "VARCHAR2",
70. "charsetName": null,
71. "length": 255,
72. "scale": null,
73. "position": 3,
74. "optional": false,
75. "autoIncremented": false,
76. "generated": false
77. },
78. {
79. "name": "EMAIL",
80. "jdbcType": 12,
81. "nativeType": null,
82. "typeName": "VARCHAR2",
83. "typeExpression": "VARCHAR2",
84. "charsetName": null,
85. "length": 255,
86. "scale": null,
87. "position": 4,
88. "optional": false,
89. "autoIncremented": false,
90. "generated": false
91. }
92. ]
93. }
94. }
95. ]
96. }
97. }

Table 4. Descriptions of fields in messages emitted to the schema change topic

| Item | Field name | Description |

| --- | --- | --- | | 1 | databaseName schemaName | Identifies the database and the schema that contains the change. | | 2 | ddl | This field contains the DDL that is responsible for the schema change. | | 3 | tableChanges | An array of one or more items that contain the schema changes generated by a DDL command. | | 4 | type | Describes the kind of change. The type can be set to one of the following values: CREATE Table created. ALTER Table modified. DROP Table deleted. | | 5 | id | Full identifier of the table that was created, altered, or dropped. In the case of a table rename, this identifier is a concatenation of <old>,<new> table names. | | 6 | table | Represents table metadata after the applied change. | | 7 | primaryKeyColumnNames | List of columns that compose the table’s primary key. | | 8 | columns | Metadata for each column in the changed table. |

In messages that the connector sends to the schema change topic, the message key is the name of the database that contains the schema change. In the following example, the payload field contains the databaseName key:

1. {
2. "schema": {
3. "type": "struct",
4. "fields": [
5. {
6. "type": "string",
7. "optional": false,
8. "field": "databaseName"
9. }
10. ],
11. "optional": false,
12. "name": "io.debezium.connector.oracle.SchemaChangeKey"
13. },
14. "payload": {
15. "databaseName": "ORCLPDB1"
16. }
17. }

Transaction Metadata

Debezium can generate events that represent transaction metadata boundaries and that enrich data change event messages.

Database transactions are represented by a statement block that is enclosed between the BEGIN and END keywords. Debezium generates transaction boundary events for the BEGIN and END delimiters in every transaction. Transaction boundary events contain the following fields:

status

BEGIN or END

id

String representation of unique transaction identifier.

event_count (for END events)

Total number of events emmitted by the transaction.

data_collections (for END events)

An array of pairs of data_collection and event_count elements that indicates number of events that the connector emits for changes that originate from a data collection.

The following example shows a typical transaction boundary message:

Example: Oracle connector transaction boundary event

1. {
2. "status": "BEGIN",
3. "id": "5.6.641",
4. "event_count": null,
5. "data_collections": null
6. }

8. {
9. "status": "END",
10. "id": "5.6.641",
11. "event_count": 2,
12. "data_collections": [
13. {
14. "data_collection": "ORCLPDB1.DEBEZIUM.CUSTOMER",
15. "event_count": 1
16. },
17. {
18. "data_collection": "ORCLPDB1.DEBEZIUM.ORDER",
19. "event_count": 1
20. }
21. ]
22. }

Unless overridden via the transaction.topic option, the connector emits transaction events to the .transaction topic.

Change data event enrichment

When transaction metadata is enabled, the data message Envelope is enriched with a new transaction field. This field provides information about every event in the form of a composite of fields:

id

String representation of unique transaction identifier.

total_order

The absolute position of the event among all events generated by the transaction.

data_collection_order

The per-data collection position of the event among all events that were emitted by the transaction.

The following example shows a typical transaction event message:

1. {
2. "before": null,
3. "after": {
4. "pk": "2",
5. "aa": "1"
6. },
7. "source": {
8. ...
9. },
10. "op": "c",
11. "ts_ms": "1580390884335",
12. "transaction": {
13. "id": "5.6.641",
14. "total_order": "1",
15. "data_collection_order": "1"
16. }
17. }

Event buffering

Oracle writes all changes to the redo logs in the order in which they occur, including changes that are later discarded by a rollback. As a result, concurrent changes from separate transactions are intertwined. When the connector first reads the stream of changes, because it cannot immediately determine which changes are committed or rolled back, it temporarily stores the change events in an internal buffer. After a change is committed, the connector writes the change event from the buffer to Kafka. The connector drops change events that are discarded by a rollback.

You can configure the buffering mechanism that the connector uses by setting the property log.mining.buffer.type.

Heap

The default buffer type is configured using memory. Under the default memory setting, the connector uses the heap memory of the JVM process to allocate and manage buffered event records. If you use the memory buffer setting, be sure that the amount of memory that you allocate to the Java process can accommodate long-running and large transactions in your environment.

Infinispan

The Debezium Oracle connector can also be configured to use Infinispan as its cache provider, supporting cache stores both locally with embedded mode or remotely on a server cluster. In order to use Infinispan, the log.mining.buffer.type must be configured using either infinispan_embedded or infinispan_remote.

In order to allow flexibility with Infinispan cache configurations, the connector expects a series of cache configuration properties to be supplied when using Infinispan to buffer event data. See the configuration properties in the log.mining.buffer.infinispan.cache namespace. The contents of these configuration properties depend on whether the connector is to integrate with a remote Infinispan cluster or to use the embedded engine.

For example, the following illustrates what an embedded configuration would look like for the transaction cache property when using Infinispan in embedded mode:

1. <local-cache name="transactions">
2. <persistence passivation="false">
3. <file-store fetch-state="true" read-only="false" preload="true" shared="false" segmented="false" path="./data" />
4. </persistence>
5. </local-cache>

Looking at the configuration in-depth, the cache is configured to be persistent. All caches should be configured this way to avoid loss of transaction events across connector restarts if a transaction is in-progress. Additionally, the location where the cache is kept is defined by the path attribute and this should be a shared location accessible all possible runtime environments.

Another example, the following illustrates the same cache configured with an Infinispan cluster:

1. <distributed-cache name="transactions" statistics="true">
2. <encoding media-type="application/x-protostream" />
3. <persistence passivation="false">
4. <file-store fetch-state="true" read-only="false" preload="true" shared="false" segmented="false" />
5. </persistence>
6. </distributed-cache>

Just like the embedded local-cache configuration from the previous example, this configuration is also defined to be persistent. All caches should be configured this way to avoid loss of transaction events across connector restarts if a transaction is in-progress.

However, there are a few differences with noting. First, the cache is defined as a distributed cache rather than a local-cache. Secondly, the cache is defined to use the application/x-protostream encoding, which is required for all Debezium caches. And lastly, no path attribute is necessary on the file store definition since the Infinispan cluster will handle this automatically.

Infinispan Hotrod client integration

The Debezium Oracle connector utilizes the Hotrod client to communicate with the Infinispan cluster. Any connector property that is prefixed with log.mining.buffer.infinispan.client. will be passed directly to the Hotrod client using the infinispan.client. namespace, allowing for complete customization of how the client is to interact with the cluster.

There is at least one required configuration property that must be supplied when using this Infinspan mode:

log.mining.buffer.infinispan.client.hotrod.server_list

Specifies the list of Infinispan server hostname and port combinations, using <hostname>:<port> format.

SCN gap detection

When the Debezium Oracle connector is configured to use LogMiner, it collects change events from Oracle by using a start and end range that is based on system change numbers (SCNs). The connector manages this range automatically, increasing or decreasing the range depending on whether the connector is able to stream changes in near real-time, or must process a backlog of changes due to the volume of large or bulk transactions in the database.

Under certain circumstances, the Oracle database advances the SCN by an unusually high amount, rather than increasing the SCN value at a constant rate. Such a jump in the SCN value can occur because of the way that a particular integration interacts with the database, or as a result of events such as hot backups.

The Debezium Oracle connector relies on the following configuration properties to detect the SCN gap and adjust the mining range.

log.mining.scn.gap.detection.gap.size.min

Specifies the minimum gap size.

log.mining.scn.gap.detection.time.interval.max.ms

Specifies the maximum time interval.

The connector first compares the difference in the number of changes between the current SCN and the highest SCN in the current mining range. If the difference between the current SCN value and the highest SCN value is greater than the minimum gap size, then the connector has potentially detected a SCN gap. To confirm whether a gap exists, the connector next compares the timestamps of the current SCN and the SCN at the end of the previous mining range. If the difference between the timestamps is less than the maximum time interval, then the existence of an SCN gap is confirmed.

When an SCN gap occurs, the Debezium connector automatically uses the current SCN as the end point for the range of the current mining session. This allows the connector to quickly catch up to the real-time events without mining smaller ranges in between that return no changes because the SCN value was increased by an unexpectedly large number. When the connector performs the preceding steps in response to an SCN gap, it ignores the value that is specified by the log.mining.batch.size.max property. After the connector finishes the mining session and catches back up to real-time events, it resumes enforcement of the maximum log mining batch size.

Low change frequency offset management

The Debezium Oracle connector tracks system change numbers in the connector offsets so that when the connector is restarted, it can begin where it left off. These offsets are part of each emitted change event; however, when the frequency of database changes are low (every few hours or days), the offsets can become stale and prevent the connector from successfully restarting if the system change number is no longer available in the transaction logs.

For connectors that use non-CDB mode to connect to Oracle, you can enable heartbeat.interval.ms to force the connector to emit a heartbeat event at regular intervals so that offsets remain synchronized.

For connectors that use CDB mode to connect to Oracle, maintaining synchronization is more complicated. Not only must you set heartbeat.interval.ms, but it’s also necessary to set heartbeat.action.query. Specifying both properties is required, because in CDB mode, the connector specifically tracks changes inside the PDB only. A supplementary mechanism is needed to trigger change events from within the pluggable database. At regular intervals, the heartbeat action query causes the connector to insert a new table row, or update an existing row in the pluggable database. Debezium detects the table changes and emits change events for them, ensuring that offsets remain synchronized, even in pluggable databases that process changes infrequently.

Data change events

Every data change event that the Oracle connector emits has a key and a value. The structures of the key and value depend on the table from which the change events originate. For information about how Debezium constructs topic names, see Topic names.

Debezium and Kafka Connect are designed around continuous streams of event messages. However, the structure of these events might change over time, which can be difficult for topic consumers to handle. To facilitate the processing of mutable event structures, each event in Kafka Connect is self-contained. Every message key and value has two parts: a schema and payload. The schema describes the structure of the payload, while the payload contains the actual data.

Change event keys

For each changed table, the change event key is structured such that a field exists for each column in the primary key (or unique key constraint) of the table at the time when the event is created.

For example, a customers table that is defined in the inventory database schema, might have the following change event key:

1. CREATE TABLE customers (
2. id NUMBER(9) GENERATED BY DEFAULT ON NULL AS IDENTITY (START WITH 1001) NOT NULL PRIMARY KEY,
3. first_name VARCHAR2(255) NOT NULL,
4. last_name VARCHAR2(255) NOT NULL,
5. email VARCHAR2(255) NOT NULL UNIQUE
6. );

If the value of the .transaction configuration property is set to server1, the JSON representation for every change event that occurs in the customers table in the database features the following key structure:

1. {
2. "schema": {
3. "type": "struct",
4. "fields": [
5. {
6. "type": "int32",
7. "optional": false,
8. "field": "ID"
9. }
10. ],
11. "optional": false,
12. "name": "server1.INVENTORY.CUSTOMERS.Key"
13. },
14. "payload": {
15. "ID": 1004
16. }
17. }

The schema portion of the key contains a Kafka Connect schema that describes the content of the key portion. In the preceding example, the payload value is not optional, the structure is defined by a schema named server1.DEBEZIUM.CUSTOMERS.Key, and there is one required field named id of type int32. The value of the key’s payload field indicates that it is indeed a structure (which in JSON is just an object) with a single id field, whose value is 1004.

Therefore, you can interpret this key as describing the row in the inventory.customers table (output from the connector named server1) whose id primary key column had a value of 1004.

Change event values

The structure of a value in a change event message mirrors the structure of the message key in the change event in the message, and contains both a schema section and a payload section.

Payload of a change event value

An envelope structure in the payload sections of a change event value contains the following fields:

op

A mandatory field that contains a string value describing the type of operation. The op field in the payload of an Oracle connector change event value contains one of the following values: c (create or insert), u (update), d (delete), or r (read, which indicates a snapshot).

before

An optional field that, if present, describes the state of the row before the event occurred. The structure is described by the server1.INVENTORY.CUSTOMERS.Value Kafka Connect schema, which the server1 connector uses for all rows in the inventory.customers table.

after

An optional field that, if present, contains the state of a row after a change occurs. The structure is described by the same server1.INVENTORY.CUSTOMERS.Value Kafka Connect schema that is used for the before field.

source

A mandatory field that contains a structure that describes the source metadata for the event. In the case of the Oracle connector, the structure includes the following fields:

• The Debezium version.
• The connector name.
• Whether the event is part of an ongoing snapshot or not.
• The transaction id (not includes for snapshots).
• The SCN of the change.
• A timestamp that indicates when the record in the source data

（本文超出站点单文档体积上限，已截断）