Kafka-connect-jdbc: Multiple tasks in kafka-connect-jdbc sink causing deadlock

[kalbmj]

After increasing the "tasks.max": "1" to "tasks.max":"3" in the io.confluent.connect.jdbc.JdbcSinkConnector
we are seeing "Caused by: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction " in the kafkaconnect logs. Is this due to the sink database configurations, or does something else in kafkaconnect need to be set?

Thanks in advance.

[chazv12]

Did you find a fix? Having similar issues here when using more than one task.

[rhauch]

There should be nothing in Connect that would prevent this, other than simply have too many threads. Several DBMSes can limit the number of concurrent JDBC connections that are allowed to proceed, and often in these cases the JDBC connections just block. In either case, you should be able to get a stack dump and look where threads a blocked.

[rhauch]

Also, how many topic partitions are you consuming? You can only have as many consumers as you have topic partitions, so that could also be a reason why the sink connector tasks appear blocked: they might not have any topic partitions to consume.

[chazv12]

We have 8 partitions on the topic, I've run various combos of tasks, 2, 4, 8...all of them eventually start running into this error. 1 Task works just fine - but we're currently looking at about 30k inserts/min, which is quite slow compared to the rest of our pipeline.

I'm Using the Teradata JDBC Driver.

Here's a large snippet from the stack dump:

java.sql.SQLException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1339] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.
java.sql.SQLException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1339] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.
    ... 12 more
[2018-06-05 15:33:31,341] WARN WorkerSinkTask{id=teradata-sink-0} Commit of offsets timed out (org.apache.kafka.connect.runtime.WorkerSinkTask:191)
[2018-06-05 15:33:32,354] INFO Checking table:TABLE_NAME exists for product:Teradata schema:null catalog: (io.confluent.connect.jdbc.sink.DbMetadataQueries:50)
[2018-06-05 15:33:32,377] INFO product:Teradata schema:null catalog: -- table:TABLE_NAME is present (io.confluent.connect.jdbc.sink.DbMetadataQueries:60)
[2018-06-05 15:33:32,379] INFO Querying column metadata for product:Teradata schema:null catalog: table:TABLE_NAME (io.confluent.connect.jdbc.sink.DbMetadataQueries:84)
[2018-06-05 15:33:48,295] WARN Write of 130 records failed, remainingRetries=5 (io.confluent.connect.jdbc.sink.JdbcSinkTask:71)
java.sql.BatchUpdateException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1338] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. Details of the failure can be found in the exception chain that is accessible with getNextException.
    at com.teradata.jdbc.jdbc_4.util.ErrorFactory.makeBatchUpdateException(ErrorFactory.java:149)
    at com.teradata.jdbc.jdbc_4.util.ErrorFactory.makeBatchUpdateException(ErrorFactory.java:138)
    at com.teradata.jdbc.jdbc_4.TDPreparedStatement.executeBatchDMLArray(TDPreparedStatement.java:270)
    at com.teradata.jdbc.jdbc_4.TDPreparedStatement.executeBatch(TDPreparedStatement.java:2659)
    at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:120)
    at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:69)
    at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:69)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:495)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:288)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.sql.SQLException: [Teradata Database] [TeraJDBC 15.10.00.36] [Error 2631] [SQLState 40001] Transaction ABORTed due to deadlock.
    at com.teradata.jdbc.jdbc_4.util.ErrorFactory.cloneJDBCException(ErrorFactory.java:171)
    at com.teradata.jdbc.jdbc_4.statemachine.PreparedBatchStatementController.handleRunException(PreparedBatchStatementController.java:95)
    at com.teradata.jdbc.jdbc_4.statemachine.StatementController.runBody(StatementController.java:145)
    at com.teradata.jdbc.jdbc_4.statemachine.PreparedBatchStatementController.run(PreparedBatchStatementController.java:58)
    at com.teradata.jdbc.jdbc_4.TDStatement.executeStatement(TDStatement.java:389)
    at com.teradata.jdbc.jdbc_4.TDPreparedStatement.executeBatchDMLArray(TDPreparedStatement.java:250)
    ... 15 more
[2018-06-05 15:33:48,323] INFO Initializing writer using SQL dialect: GenericDialect (io.confluent.connect.jdbc.sink.JdbcSinkTask:52)
[2018-06-05 15:33:48,323] ERROR WorkerSinkTask{id=teradata-sink-7} RetriableException from SinkTask: (org.apache.kafka.connect.runtime.WorkerSinkTask:508)
org.apache.kafka.connect.errors.RetriableException: java.sql.SQLException: java.sql.BatchUpdateException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1338] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. Details of the failure can be found in the exception chain that is accessible with getNextException.
java.sql.SQLException: [Teradata Database] [TeraJDBC 15.10.00.36] [Error 2631] [SQLState 40001] Transaction ABORTed due to deadlock.
java.sql.SQLException: [Teradata Database] [TeraJDBC 15.10.00.36] [Error 2631] [SQLState 40001] Transaction ABORTed due to deadlock.
java.sql.SQLException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1339] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.
java.sql.SQLException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1339] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.
java.sql.SQLException: [Teradata JDBC Driver] [TeraJDBC 15.10.00.36] [Error 1339] [SQLState HY000] A failure occurred while executing a PreparedStatement batch request. The parameter set was not executed and should be resubmitted individually using the PreparedStatement executeUpdate method.

[rhauch]

The sink connector uses a single consumer group to consume all of the partitions in the specified topics, and every sink connector task that starts up joins the group, and the consumer group protocol decides which topic partitions are assigned to each task.

Each JDBC sink connector task then consumes a batch of records and uses a SQL transaction to write those records to the affected tables. It sounds like your connections are competing as concurrent transactions for the same resources. Have you tried reducing the batch size or tuning the Connect workers' consumer.* properties, like fetch.max.bytes, max.poll.interval.ms, or max.poll.records?

[chazv12]

Appreciate the help @rhauch.

I understand that increasing the number of tasks essentially adds more consumers to the consumer group.

I've been playing around with those parameters in my worker.properties file, pre-fixing them with consumer. . Just can't find a combination that seems to actually improve performance!

Maybe I'm misunderstanding how to tune the parameters.

A couple of different scenarios I've tried:

• a large consumer.fetch.max.bytes (Up to 0.5 MiB), large max.poll.records (10k) and larger consumer.fetch.wait.max.ms (1000ms) to encourage our batch inserts to be of a decent size, and lessen the IO on the database.

• Or a smaller consumer.fetch.wait.max.ms (I set to as low as 100ms) to as much data as we can get as fast as we can to each consumer. Kept the other parameters large. Though I could see the more frequent polls causing more IO deadlock issues on the database.

A single task running these parameters seemed to perform at the same level (~30k inserts/min), while multiple tasks running start to error after a couple of minutes.

Any suggestions on the approach you would take in tuning these parameters?

[jeffbeagley]

Any updates on this? I'm seeing this as well with 10 tasks consuming 10 partitions on the same topic, all writing to the same table. I fear changing the isolation level may cause events to be sent out of order (using CDC).

[sunilkpokkalla]

I am facing the below error

(io.confluent.connect.jdbc.sink.JdbcSinkConfig:223)
[2019-06-12 21:37:49,625] INFO Initializing writer using SQL dialect: GenericDialect (io.confluent.connect.jdbc.sink.JdbcSinkTask:52)
[2019-06-12 21:37:49,626] INFO WorkerSinkTask{id=kafka-transferlabresultacc-tdsink-0} Sink task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSinkTask:268)
[2019-06-12 21:37:50,077] INFO [Consumer clientId=kafka-transferlabresultacc-tdsink, groupId=connect-kafka-transferlabresultacc-tdsink] Discovered coordinator cilhdkfs0302.sys.cigna.com:9093 (id: 2147483551 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:341)
[2019-06-12 21:37:50,085] INFO [Consumer clientId=kafka-transferlabresultacc-tdsink, groupId=connect-kafka-transferlabresultacc-tdsink] Revoking previously assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:341)
[2019-06-12 21:37:50,085] INFO [Consumer clientId=kafka-transferlabresultacc-tdsink, groupId=connect-kafka-transferlabresultacc-tdsink] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:336)
[2019-06-12 21:37:53,161] INFO [Consumer clientId=kafka-transferlabresultacc-tdsink, groupId=connect-kafka-transferlabresultacc-tdsink] Successfully joined group with generation 3 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:341)
[2019-06-12 21:37:53,164] INFO [Consumer clientId=kafka-transferlabresultacc-tdsink, groupId=connect-kafka-transferlabresultacc-tdsink] Setting newly assigned partitions [cdi-kafka-transferlabresultacc-0, cdi-kafka-transferlabresultacc-1, cdi-kafka-transferlabresultacc-2, cdi-kafka-transferlabresultacc-3] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:341)
[2019-06-12 21:37:54,700] INFO Checking table:TransferLabResultAcc1 exists for product:Teradata schema:null catalog: (io.confluent.connect.jdbc.sink.DbMetadataQueries:50)
[2019-06-12 21:37:54,899] INFO product:Teradata schema:null catalog: -- table:TransferLabResultAcc1 is absent (io.confluent.connect.jdbc.sink.DbMetadataQueries:60)
[2019-06-12 21:37:54,939] ERROR WorkerSinkTask{id=kafka-transferlabresultacc-tdsink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. (org.apache.kafka.connect.runtime.WorkerSinkTask:515)
java.lang.UnsupportedOperationException
at io.confluent.connect.jdbc.sink.dialect.GenericDialect.getCreateQuery(GenericDialect.java:34)
at io.confluent.connect.jdbc.sink.DbStructure.create(DbStructure.java:90)
at io.confluent.connect.jdbc.sink.DbStructure.createOrAmendIfNecessary(DbStructure.java:63)
at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:78)
at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:66)
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:69)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:495)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:288)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
[2019-06-12 21:37:54,940] ERROR WorkerSinkTask{id=kafka-transferlabresultacc-tdsink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:517)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:288)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.UnsupportedOperationException
at io.confluent.connect.jdbc.sink.dialect.GenericDialect.getCreateQuery(GenericDialect.java:34)
at io.confluent.connect.jdbc.sink.DbStructure.create(DbStructure.java:90)
at io.confluent.connect.jdbc.sink.DbStructure.createOrAmendIfNecessary(DbStructure.java:63)
at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:78)
at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:66)
at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:69)
at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:495)
... 10 more
[2019-06-12 21:37:54,941] ERROR WorkerSinkTask{id=kafka-transferlabresultacc-tdsink-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:173)
[2019-06-12 21:37:54,941] INFO Stopping task (io.confluent.connect.jdbc.sink.JdbcSinkTask:96)

[woodlee]

If the DB deadlocks are over row-level locks, check to make sure that the key being used for the source Kafka records is 1:1 with the primary key of the target DB table. Otherwise you might have two Connect tasks, consuming records from two different Kafka partitions, trying to update the same target DB row concurrently.

[imranrajjad]

we are experiencing the same row lock problem on Oracle with a topic which is receiving 4k messages per second. Some highlights of the sink configuration are:

number of tasks = number of partitions on input topic
batch size = 1000
mode : UPSERT
Connect versions tried so far ( 4.1.3 and 5.5.0)

Further using Kafkacat it has been verified that incoming messages have keys and they don't land across different partitions. However, a small analysis of Connect logs suggested that messages with similar keys do get picked up by different tasks (specially when the task count is greater than partition count).

Other things tried:

• We have also tried changing the work consumer properties consumer.max.poll.records=1000.By increasing the poll records to 1000 we were hoping to have a bigger batch which could have similar keys accumulated in same batch but it made things worse.
• Using a single task to induce synchronous behavior, the row locks still appeared.

Some questions:

• Are there any JDBC tuning guidelines that take into account the throughput, network latency etc?
• Is it possible to make the JDBC Sink run in Synchronous mode?
• Do we need to slow down the JDBC Sink?
• Will a smaller batch size help, so that records can be rid off as quickly as possible?