JDBC Sink 连接器抛出 java.sql.BatchUpdateException答案

【问题标题】：JDBC Sink connector throwing java.sql.BatchUpdateExceptionJDBC Sink 连接器抛出 java.sql.BatchUpdateException
【发布时间】：2019-11-09 09:03:20
【问题描述】：

几周前我启动了一个 Sink JDBC。一切都很好，直到日志开始出现此错误：

[2019-06-27 11:35:44,121] WARN Write of 500 records failed, remainingRetries=10 (io.confluent.connect.jdbc.sink.JdbcSinkTask:68)
java.sql.BatchUpdateException: [Teradata JDBC Driver] [TeraJDBC 16.20.00.10] [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:276)
        at com.teradata.jdbc.jdbc_4.TDPreparedStatement.executeBatch(TDPreparedStatement.java:2754)
        at io.confluent.connect.jdbc.sink.BufferedRecords.flush(BufferedRecords.java:99)
        at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:78)
        at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:62)
        at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:66)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:429)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
        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:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

我已经尝试过降低 batch.size 属性，甚至低至 100，但仍然失败。

添加连接器状态：

{"name":"teradata-sink-K_C_OSUSR_DGL_DFORM_I1-V2",
"connector":{
"state":"RUNNING",
"worker_id":"10.28.148.64:41029"},
"tasks":[{"state":"FAILED","trace":"org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
 org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:451)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
 org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
 org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
 org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
 java.util.concurrent.FutureTask.run(FutureTask.java:266)
 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 java.lang.Thread.run(Thread.java:748)\n","id":0,"worker_id":"10.28.148.64:41029"}]}

【问题讨论】：

您的连接器的状态如何？ curl localhost:8083/connectors/connector-name/status | jq
@GiorgosMyrianthous 我将其添加到问题中
您确定您的批次不包含对相同主键的更新吗？这可能是某些数据库的问题。
@MarkusRother，不，我很确定，因为我们之前遇到过这个问题，现在没有发生。

标签： apache-kafka teradata apache-kafka-connect

【解决方案1】：

我在尝试使用 JDBC 从 spark 写入 teradata 时遇到了类似的问题。显然，这仅发生在 teradata 上。当您尝试使用多个 JDBC 连接并行写入一个表时，就会发生这种情况。就我而言，spark 在将数据写入 Teradata 时将多个 jdbc 连接分叉到 teradata。您的选择是：

您必须限制您的应用程序不要创建多个 jdbc 连接，如果您有访问权限，您可以从 Viewpoint 监控它。
您可以尝试提供一个 jdbc type=FASTLOAD 并查看它是否正常工作。 like this
或者去 TPT 内部处理这个问题。

【讨论】：