Kafka Connector stops if connection with database breaks with Source Connector

[y24jds]

Scenario: Kafka-connect with JDBC Source connector talking to SQLServer is up and running and polling a database table. The SQLServer is taken down. The Kafka-connect log file shows:

[2019-12-12 09:58:14,124] ERROR WorkerSourceTask{id=test-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:179) org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: I/O Error: DB server closed connection. at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:69) at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:310) at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:245) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:221) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227) 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) Caused by: java.sql.SQLException: I/O Error: DB server closed connection. at net.sourceforge.jtds.jdbc.TdsCore.executeSQL(TdsCore.java:1093) at net.sourceforge.jtds.jdbc.JtdsStatement.executeSQL(JtdsStatement.java:563) at net.sourceforge.jtds.jdbc.JtdsStatement.executeImpl(JtdsStatement.java:809) at net.sourceforge.jtds.jdbc.JtdsStatement.execute(JtdsStatement.java:1282) at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.isConnectionValid(GenericDatabaseDialect.java:243) at io.confluent.connect.jdbc.util.CachedConnectionProvider.isConnectionValid(CachedConnectionProvider.java:79) at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:63) ... 10 more Suppressed: java.sql.SQLException: Invalid state, the Connection object is closed. at net.sourceforge.jtds.jdbc.TdsCore.checkOpen(TdsCore.java:481) at net.sourceforge.jtds.jdbc.TdsCore.clearResponseQueue(TdsCore.java:767) at net.sourceforge.jtds.jdbc.JtdsStatement.reset(JtdsStatement.java:722) at net.sourceforge.jtds.jdbc.JtdsStatement.close(JtdsStatement.java:966) at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.isConnectionValid(GenericDatabaseDialect.java:248) ... 12 more Caused by: java.io.IOException: DB server closed connection. at net.sourceforge.jtds.jdbc.SharedSocket.readPacket(SharedSocket.java:852) at net.sourceforge.jtds.jdbc.SharedSocket.getNetPacket(SharedSocket.java:731) at net.sourceforge.jtds.jdbc.ResponseStream.getPacket(ResponseStream.java:477) at net.sourceforge.jtds.jdbc.ResponseStream.read(ResponseStream.java:114) at net.sourceforge.jtds.jdbc.ResponseStream.peek(ResponseStream.java:99) at net.sourceforge.jtds.jdbc.TdsCore.wait(TdsCore.java:4127) at net.sourceforge.jtds.jdbc.TdsCore.executeSQL(TdsCore.java:1086) ... 16 more [2019-12-12 09:58:14,125] ERROR WorkerSourceTask{id=test-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:180)

If I bring the database back online, the connector stays down and the only way to get the system working again is to restart the connector task.

I tried setting the properties: connection.attempts connection.backoff.ms

But those don't seem to have any effect since kafka-connect has already decided to kill the task.

My only option is to write some external monitor that checks the connector status and if it shows down, to periodically give it a restart command. I haven't studied this blog post but it seems like this is what is being suggested: https://rmoff.net/2019/06/06/automatically-restarting-failed-kafka-connect-tasks/

Shouldn't kafka-connect have an automatic way to keep trying to reconnect to a failed database?

[gharris1727]

It appears that a fix for this has already been proposed in #715 . We will follow up with that PR to get the fix merged in.

[mdespriee]

Any news on this, now that #715 is merged ?

I have a similar problem with a jdbc sink.

[bonndan]

Any news?

[spartan-it]

I have a similar issue with an Oracle connection. According to [#715 ], it was merged. But I keep having this closed connection issue. Maybe it is an Oracle DB-specific bug. Here are my logs:

org.apache.kafka.connect.errors.DataException: java.sql.SQLRecoverableException: Closed Connection: getDouble
    at io.confluent.connect.jdbc.source.TimestampTableQuerier.doExtractRecord(TimestampTableQuerier.java:150)
    at io.confluent.connect.jdbc.source.TimestampTableQuerier.extractRecord(TimestampTableQuerier.java:130)
    at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:441)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:305)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:249)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.sql.SQLRecoverableException: Closed Connection: getDouble
    at oracle.jdbc.driver.GeneratedScrollableResultSet.getDouble(GeneratedScrollableResultSet.java:210)
    at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.lambda$columnConverterFor$25(GenericDatabaseDialect.java:1378)
    at io.confluent.connect.jdbc.source.SchemaMapping$FieldSetter.setField(SchemaMapping.java:159)
    at io.confluent.connect.jdbc.source.TimestampTableQuerier.doExtractRecord(TimestampTableQuerier.java:144)
    ... 11 more

I find it quite annoying to have to restart the connector every time this error happens. The connector should just restart on its own with the appropriate settings.