-
Bug
-
Resolution: Unresolved
-
Major
-
2.6.0.Alpha2
-
None
-
False
-
None
-
False
Bug report
For bug reports, provide this information, please:
What Debezium connector do you use and what version?
debezium-connector-oracle version 2.6.0.Alpha2
What is the connector configuration?
{
"name": "source-test-connector",
"config": {
"connector.class": "io.debezium.connector.oracle.OracleConnector",
"tasks.max": "1",
"database.hostname": "oracle",
"database.port": "1521",
"database.user": "c##dbzuser",
"database.password": "dbz",
"database.dbname": "orclcdb",
"database.pdb.name": "orclpdb1",
"database.connection.adapter": "logminer",
"topic.prefix": "dbz",
"schema.name.adjustment.mode": "avro",
"table.include.list": "C##DBZUSER.TEST_TABLE",
"include.schema.changes": "false",
"schema.history.internal.kafka.bootstrap.servers" : "kafka:9092",
"schema.history.internal.kafka.topic": "schema-changes.test",
"heartbeat.interval.ms": "60000",
"log.mining.strategy": "online_catalog",
"log.mining.query.filter.mode": "in",
"custom.metric.tags": "connector=source-test-connector",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"
}
}
What is the captured database version and mode of deployment?
Oracle Database 19, Docker
What behaviour do you expect?
Only the last event that occurred within a transaction must contain the commit SCN of the last transaction in offset part:
record 1:
- offset part:
"commit_scn" = "2375324:1:0500140048030000" <- commit scn previous transaction "snapshot_scn" = "2375217" "scn" = "2375307" <- smallest scn from transaction buffer
- source part:
"scn" = "2375308" <- event scn "commit_scn" = "2375338" <- commit scn current transaction
record 2:
- offset part:
"commit_scn" = "2375338:1:0200140004020000" <- commit scn current transaction "snapshot_scn" = "2375217" "scn" = "2375307" <- smallest scn from transaction buffer
- source part:
"scn" = "2375337" <- event scn "commit_scn" = "2375338" <- commit scn current transaction
What behaviour do you see?
Let's imagine a situation in Oracle database: multiple DML events occurred within the same transaction, which were handled by Oracle connector and sent to Kafka in different polling iterations. Suppose that while sending messages to Kafka, only the first event was sent successfully, and sending of the second event failed. Then the connector restarted for some reason and the second event would never be sent to Kafka again.
This can happen because all events contain the same commit SCN of the last transaction in offset part:
record 1:
- offset part:
"commit_scn" = "2375338:1:0200140004020000" <- commit scn current transaction "snapshot_scn" = "2375217" "scn" = "2375307" <- smallest scn from transaction buffer
- source part:
"scn" = "2375308" <- event scn "commit_scn" = "2375338" <- commit scn current transaction
record 2:
- offset part:
"commit_scn" = "2375338:1:0200140004020000" <- commit scn current transaction "snapshot_scn" = "2375217" "scn" = "2375307" <- smallest scn from transaction buffer
- source part:
"scn" = "2375337" <- event scn "commit_scn" = "2375338" <- commit scn current transaction
Do you see the same behaviour using the latest relesead Debezium version?
Yes
How to reproduce the issue using our tutorial deployment?
1. Create a new table:
CREATE TABLE c##dbzuser.test_table ( id NUMBER(10) NOT NULL PRIMARY KEY, text VARCHAR2(300) );
2. Create a new connector:
curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors -d '
{
"name": "source-test-connector",
"config": {
"connector.class": "io.debezium.connector.oracle.OracleConnector",
"tasks.max": "1",
"database.hostname": "oracle",
"database.port": "1521",
"database.user": "c##dbzuser",
"database.password": "dbz",
"database.dbname": "orclcdb",
"database.pdb.name": "orclpdb1",
"database.connection.adapter": "logminer",
"topic.prefix": "dbz",
"schema.name.adjustment.mode": "avro",
"table.include.list": "C##DBZUSER.TEST_TABLE",
"include.schema.changes": "false",
"schema.history.internal.kafka.bootstrap.servers" : "kafka:9092",
"schema.history.internal.kafka.topic": "schema-changes.test",
"heartbeat.interval.ms": "60000",
"log.mining.strategy": "online_catalog",
"log.mining.query.filter.mode": "in",
"custom.metric.tags": "connector=source-test-connector",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"
}
}'
3. Insert a new record into the table in a loop:
INSERT INTO c##dbzuser.test_table (id, text) VALUES (1, 'text'); INSERT INTO c##dbzuser.test_table (id, text) VALUES (2, 'text'); commit;
4. Check offset part in records sent to Kafka.
- duplicates
-
-
- Coding In Progress
-
