Debezium JDBC sinc-connector error deadlock

## Bug report

**What Debezium connector do you use and what version?**

`quay/debezium/connect:3.3.1.Final`

---

**What is the connector configuration?**

source
{
"name": "{{sourceConnectorName}}",
"config": {
"connector.class": "io.debezium.connector.oracle.OracleConnector",
"tasks.max": "1",
"database.hostname": "{{sourceDatabaseHost}}",
"database.port": "{{sourceDatabasePort}}",
"database.user": "{{sourceDatabaseUser}}",
"database.password": "{{sourceDatabasePassword}}",
"database.dbname": "{{sourceDatabaseName}}",
"table.include.list": "SCHEMA.DEBEZIUM_SIGNAL,SCHEMA2.TABLE",
"column.include.list": "SCHEMA\\.DEBEZIUM_SIGNAL\\.(ID|TYPE|DATA),SCHEMA2\\.TABLE\\.(ID|ID_OP|ID_ART|LASTDATE)",
"topic.prefix": "{{topicPrefix}}",
"database.server.name": "{{topicPrefix}}",
"schema.history.internal.kafka.topic": "dbz_oracle_wpms_history",
"schema.history.internal.kafka.bootstrap.servers": "{{kafkaBootstrapServers}}",

"message.key.columns": "SCHEMA.DEBEZIUM_SIGNAL:ID;SHEMA2.TABLE:ID,LASTDATE;",
"schema.include.list": "SCHEMA,SCHEMA2",

"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
"key.converter.apicurio.registry.url": "{{apicurioRegistryUrl}}",
"key.converter.apicurio.registry.auto-register": "true",
"key.converter.apicurio.registry.find-latest": "true",
"key.converter.schemas.enable": "false",
"key.converter.apicurio.registry.headers.enabled": "false",
"key.converter.apicurio.registry.as-confluent": "true",
"key.converter.apicurio.use-id": "contentId",
"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
"value.converter.apicurio.registry.url": "{{apicurioRegistryUrl}}",
"value.converter.apicurio.registry.auto-register": "true",
"value.converter.apicurio.registry.find-latest": "true",
"value.converter.schemas.enable": "false",
"value.converter.apicurio.registry.headers.enabled": "false",
"value.converter.apicurio.registry.as-confluent": "true",
"value.converter.apicurio.use-id": "contentId",
"schema.name.adjustment.mode": "avro",

"header.converter": "org.apache.kafka.connect.json.JsonConverter",
"header.converter.schemas.enable": "true",

"signal.enable.channels": "source",
"signal.data.collection": "RCDB.WPMS.DEBEZIUM_SIGNAL",

"topic.creation.enable": "true",
"topic.creation.default.replication.factor": 1,
"topic.creation.default.partitions": 5,
"topic.creation.default.retention.ms": 345600000,
"topic.creation.default.cleanup.policy": "delete",

"log.mining.strategy": "hybrid",
"log.mining.query.filter.mode": "in",
"log.mining.transaction.retention.ms": 900000,
"log.mining.batch.size.max": 20000000,
"log.mining.batch.size.default": 5000000,
"log.mining.batch.size.increment": 1000000,
"log.cleanup.policy": "delete",
"log.retention.ms": 345600000,

"poll.interval.ms": 5,

"incremental.snapshot.chunk.size": 50000,
"incremental.snapshot.allow.schema.changes": "true",
"snapshot.fetch.size": 50000,

"snapshot.mode": "no_data",
"schema.history.internal.store.only.captured.tables.ddl": "true",
"snapshot.database.errors.max.retries": 2,
"internal.log.mining.log.query.max.retries": 15,

"heartbeat.interval.ms": "10000",
"heartbeat.action.query": "MERGE INTO SCHEMA.DEBEZIUM_HEARTBEAT t USING (SELECT 1 id, CURRENT_TIMESTAMP ts FROM dual) s ON (t.id = s.id) WHEN MATCHED THEN UPDATE SET t.ts = s.ts WHEN NOT MATCHED THEN INSERT (id, ts) VALUES (s.id, s.ts)",

"notification.enabled.channels": "sink,jmx,log",
"notification.sink.topic.name": "debezium_notifications"
}
}

SINK CONNECTOR
{
"name": "{{targetConnectorName}}",
"config": {
"connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
"tasks.max": "5",
"connection.url": "{{targetDbUrl}}?reWriteBatchedInserts=true",
"connection.username": "{{targetDbUsername}}",
"connection.password": "{{targetDbPassword}}",
"topics.regex": "{{topicPrefix}}.SCHEMA2.TABLE",
"table.name.format": "${source.schema}.${source.table}",

"delete.enabled": "true",
"primary.key.mode": "record_key",
"primary.key.fields": "LASTDATE,ID",
"insert.mode": "upsert",

"key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
"key.converter.apicurio.registry.url": "{{apicurioRegistryUrl}}",
"key.converter.apicurio.registry.auto-register": "true",
"key.converter.apicurio.registry.find-latest": "true",
"key.converter.schemas.enable": "false",
"key.converter.apicurio.registry.headers.enabled": "false",
"key.converter.apicurio.registry.as-confluent": "true",
"key.converter.apicurio.use-id": "contentId",
"value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
"value.converter.apicurio.registry.url": "{{apicurioRegistryUrl}}",
"value.converter.apicurio.registry.auto-register": "true",
"value.converter.apicurio.registry.find-latest": "true",
"value.converter.schemas.enable": "false",
"value.converter.apicurio.registry.headers.enabled": "false",
"value.converter.apicurio.registry.as-confluent": "true",
"value.converter.apicurio.use-id": "contentId",

"schema.name.adjustment.mode": "avro",

"header.converter": "org.apache.kafka.connect.json.JsonConverter",
"header.converter.schemas.enable": "true",
"use.reduction.buffer": "true"
}
}
---

**What is the captured database version and mode of deployment?**

_E.g. on-premises, with a specific cloud provider, etc._

`Oracle RAC 19.0.0.0.0`

---

**What behavior do you expect?**

With the correct configuration of the connector for replicating data from Oracle to PostgreSQL via Kafka, we expect:

* **Stable Data Flow**: Changes from the source Oracle tables are applied to the corresponding PostgreSQL tables continuously and with minimal latency.
* **High Performance**: Efficient handling of high load and rapid reduction of consumer lag is achieved by leveraging batch processing (`batch.size > 1`) and parallel tasks (`tasks.max > 1`).
* **Fault Tolerance**: In case of transient network or database errors, the connector should automatically retry operations (`flush.max.retries`) without data loss and resume normal operation once the cause is resolved.
* **Atomicity and Consistency**: All insert, update, and delete operations must execute safely under concurrent processing conditions without compromising data integrity.

---

**What behavior do you see?**

The actual operation of the connector is characterized by instability and a direct conflict between configuration goals:

* **Periodic Deadlocks**: Errors with the message `ERROR: deadlock detected` regularly appear in both PostgreSQL and connector logs. This causes individual connector tasks to fail (transition to a `FAILED` state) without automatic recovery.
* **Stability vs. Performance Conflict**:
* **Scenario A (High Performance)**: When using performance-optimized settings (`tasks.max=5`, `batch.size=50`), deadlocks occur frequently, disrupting stability.
* **Scenario B (Stability)**: Setting `batch.size=1` stops the deadlock errors, but performance drops to an unacceptable level because each record is processed in a separate transaction.
* **Ambiguous Role of Parallelism**: The deadlock problem **occurs more frequently** when `tasks.max` is increased but **does not disappear completely** even when `tasks.max=1`. This indicates that resource contention arises not only between the connector's own tasks but also between the connector and other processes in the database (e.g., the main application).
* **Root Cause (Based on Analysis)**: The problem is linked to the **`INSERT ... ON CONFLICT ... DO UPDATE` (UPSERT)** mechanism that Debezium uses by default to apply changes. When multiple records are processed concurrently—either within a single batch or by different tasks—the order in which rows are locked in PostgreSQL can lead to a classic cyclic wait (deadlock).

I looked into kafka topics. I have a lot of messages that have the same ID, but the LASTDATE changes and they end up in different partitions.

1 partition: ID = 121, LASTDATE: '2025-12-09 16:11:56'
2 partition: ID = 121, LASTDATE: '2025-12-09 16:16:23'

---

**Do you see the same behaviour using the latest released Debezium version?**

_Ideally, also verify with the latest Alpha/Beta/CR version._

`YES`

---

**Do you have the connector logs, ideally from start till finish?**

_You might be asked later to provide [DEBUG/TRACE](https://debezium.io/documentation/reference/stable/operations/logging.html) level log._

Logs with Error:
"org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\
tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:659)\n\
tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:360)\n\
tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:262)\n\
tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:226)\n\
tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:243)\n\
tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:298)\n\
tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:254)\n\tat
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)\n\tat
java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)\n\tat
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)\n\tat
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)\n\tat
java.base/java.lang.Thread.run(Thread.java:1583)\n
Caused by: org.apache.kafka.connect.errors.ConnectException: JDBC sink connector failure\n\tat
io.debezium.connector.jdbc.JdbcSinkConnectorTask.put(JdbcSinkConnectorTask.java:153)\n\tat
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:629)\n\t...
11 more\nCaused by: org.apache.kafka.connect.errors.ConnectException: Failed to flush records for table 'TABLE'\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.executeWithRetries(JdbcChangeEventSink.java:468)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.flushBufferWithRetries(JdbcChangeEventSink.java:242)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.flushBufferWithRetries(JdbcChangeEventSink.java:237)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.execute(JdbcChangeEventSink.java:154)\n\tat
io.debezium.connector.jdbc.JdbcSinkConnectorTask.put(JdbcSinkConnectorTask.java:163)\n\t...
12 more\nCaused by: org.hibernate.exception.LockAcquisitionException: Error executing work
[Batch entry 3 DELETE FROM schema.table WHERE id=('1596669153'::int8) AND lastdate=('2025-12-08 18:30:10'::timestamp)
was aborted: ERROR: deadlock detected\n Detail: Process 2674008 waits for ShareLock on transaction 153902706;
blocked by process 360094.\nProcess 360094 waits for ShareLock on transaction 153903995; blocked by process 2674008.\n
Hint: See server log for query details.\n Where: while deleting tuple (1432,229) in relation \"table_p20251208\"
Call getNextException to see other errors in the batch.] [n/a]\n\tat
org.hibernate.dialect.PostgreSQLDialect.lambda$buildSQLExceptionConversionDelegate$1(PostgreSQLDialect.java:1048)\n\tat
org.hibernate.exception.internal.StandardSQLExceptionConverter.convert(StandardSQLExceptionConverter.java:34)\n\tat
org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:115)\n\tat
org.hibernate.engine.jdbc.spi.SqlExceptionHelper.convert(SqlExceptionHelper.java:101)\n\tat
org.hibernate.engine.jdbc.internal.JdbcCoordinatorImpl.coordinateWork(JdbcCoordinatorImpl.java:319)\n\tat
org.hibernate.internal.AbstractSharedSessionContract.doWork(AbstractSharedSessionContract.java:1162)\n\tat
org.hibernate.internal.AbstractSharedSessionContract.doWork(AbstractSharedSessionContract.java:1150)\n\tat
io.debezium.connector.jdbc.RecordWriter.write(RecordWriter.java:52)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.flushBuffer(JdbcChangeEventSink.java:257)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.lambda$flushBufferWithRetries$1(JdbcChangeEventSink.java:243)\n\tat
io.debezium.connector.jdbc.JdbcChangeEventSink.executeWithRetries(JdbcChangeEventSink.java:460)\n\t...
16 more\nCaused by: java.sql.BatchUpdateException: Batch entry 3 DELETE FROM schema.table WHERE id=('1596669153'::int8)
AND lastdate=('2025-12-08 18:30:10'::timestamp) was aborted: ERROR: deadlock detected\n
Detail: Process 2674008 waits for

on transaction 153902706; blocked by process 360094.\n
Process 360094 waits for ShareLock on transaction 153903995; blocked by process 2674008.\n
Hint: See server log for query details.\n Where: while deleting tuple (1432,229) in relation \"table_p20251208\"
Call getNextException to see other errors in the batch.\n\tat
org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:165)\n\tat
org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2422)\n\tat
org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:580)\n\tat
org.postgresql.jdbc.PgStatement.internalExecuteBatch(PgStatement.java:886)\n\tat
org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:910)\n\tat
org.postgresql.jdbc.PgPreparedStatement.executeBatch(PgPreparedStatement.java:1778)\n\tat com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeBatch(NewProxyPreparedStatement.java:2544)\n\tat io.debezium.connector.jdbc.RecordWriter.lambda$processBatch$0(RecordWriter.java:90)\n\tat org.hibernate.jdbc.WorkExecutor.executeWork(WorkExecutor.java:37)\n\tat org.hibernate.internal.AbstractSharedSessionContract.lambda$doWork$8(AbstractSharedSessionContract.java:1151)\n\tat org.hibernate.engine.jdbc.internal.JdbcCoordinatorImpl.coordinateWork(JdbcCoordinatorImpl.java:314)\n\t... 22 more\nCaused by: org.postgresql.util.PSQLException: ERROR: deadlock detected\n Detail: Process 2674008 waits for ShareLock on transaction 153902706; blocked by process 360094.\nProcess 360094 waits for ShareLock on transaction 153903995; blocked by process 2674008.\n Hint: See server log for query details.\n Where: while deleting tuple (1432,229) in relation \"table_p20251208\"\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2734)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2421)\n\t... 31 more\n" ShareLock

---

**How to reproduce the issue using our [tutorial](https://github.com/debezium/debezium-examples/tree/main/tutorial) deployment?**

Create a table with 3 columns ID, LASTDATE, TEST. PR(ID, LASTDATE) Select 5 partitions. And start transferring data from ORACLE to Postgre.