search

confluentinc / kafka-connect-jdbc

Kafka Connect connector for JDBC-compatible databases
Other
16 stars 956 forks source link

WorkerSourceTask: Failed to flush, timed out while waiting for producer to flush outstanding 1 messages #1168

Open rupeshmore85 opened 2 years ago

rupeshmore85 commented 2 years ago

I am running a Kafka connector for MirrorMakerCheckpoint connector on confluent 7.0.0. However the checkpoint connector WorkerTask fails. I have tried to set offset.flush.timeout.ms = 100000 as mentioned in : https://github.com/confluentinc/kafka-connect-jdbc/issues/161#issuecomment-277154340 , however it didn't work for me. Any help would be highly appreciated.

[2022-02-09 05:29:32,015] INFO App info kafka.producer for connector-producer-mm2-rivdev_d_vcs-cpc-9 unregistered (org.apache.kafka.common.utils.AppInfoParser)
[2022-02-09 05:29:33,911] INFO WorkerSourceTask{id=mm2-rivdev_d_vcs-cpc-3} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2022-02-09 05:29:33,911] ERROR WorkerSourceTask{id=mm2-rivdev_d_vcs-cpc-3} Failed to flush, timed out while waiting for producer to flush outstanding 1 messages (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2022-02-09 05:29:33,911] WARN WorkerSourceTask{id=mm2-rivdev_d_vcs-cpc-3} After being scheduled for shutdown, the orphan task threw an uncaught exception. A newer instance of this task might be already running (org.apache.kafka.connect.runtime.WorkerTask)
java.lang.IllegalStateException: Cannot perform operation after producer has been closed
    at org.apache.kafka.clients.producer.KafkaProducer.throwIfProducerClosed(KafkaProducer.java:896)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:905)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:889)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:368)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:256)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:241)
    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)
sergiotravassos commented 1 year ago

Same problem.

I have tried too

Set offset.flush.timeout.ms = 100000 as mentioned in : https://github.com/confluentinc/kafka-connect-jdbc/issues/161#issuecomment-277154340 , however it didn't work for me.

2023-09-06 17:43:16 [2023-09-06 16:43:16,640] INFO [oracle-db-source|task-0] Started JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask:307)
2023-09-06 17:43:16 [2023-09-06 16:43:16,646] INFO [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:233)
2023-09-06 17:43:16 [2023-09-06 16:43:16,736] INFO [oracle-db-source|task-0] Begin using SQL query: SELECT ID, SC_ID, CREATE_DATE, UPDATE_DATE FROM SC.EXAMPLE WHERE "ID" > ? ORDER BY "ID" ASC (io.confluent.connect.jdbc.source.TableQuerier:182)
2023-09-06 17:43:16 [2023-09-06 16:43:16,764] WARN [oracle-db-source|task-0] [Producer clientId=connector-producer-oracle-db-source-0] Error while fetching metadata with correlation id 3 : {=INVALID_TOPIC_EXCEPTION} (org.apache.kafka.clients.NetworkClient:1073)
2023-09-06 17:43:16 [2023-09-06 16:43:16,765] ERROR [oracle-db-source|task-0] [Producer clientId=connector-producer-oracle-db-source-0] Metadata response reported invalid topics [] (org.apache.kafka.clients.Metadata:294)
2023-09-06 17:43:16 [2023-09-06 16:43:16,766] ERROR [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} failed to send record to :  (org.apache.kafka.connect.runtime.WorkerSourceTask:370)
2023-09-06 17:43:16 org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: []
2023-09-06 17:43:16 [2023-09-06 16:43:16,766] INFO [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:478)
2023-09-06 17:43:16 [2023-09-06 16:43:16,766] INFO [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} flushing 1 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:495)
2023-09-06 17:43:21 [2023-09-06 16:43:21,766] ERROR [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} Failed to flush, timed out while waiting for producer to flush outstanding 1 messages (org.apache.kafka.connect.runtime.WorkerSourceTask:500)
2023-09-06 17:43:21 [2023-09-06 16:43:21,766] ERROR [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:187)
2023-09-06 17:43:21 org.apache.kafka.connect.errors.ConnectException: Unrecoverable exception from producer send callback
2023-09-06 17:43:21     at org.apache.kafka.connect.runtime.WorkerSourceTask.maybeThrowProducerSendException(WorkerSourceTask.java:282)
2023-09-06 17:43:21     at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:336)
2023-09-06 17:43:21     at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:264)
2023-09-06 17:43:21     at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
2023-09-06 17:43:21     at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
2023-09-06 17:43:21     at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
2023-09-06 17:43:21     at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
2023-09-06 17:43:21     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
2023-09-06 17:43:21     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
2023-09-06 17:43:21     at java.base/java.lang.Thread.run(Thread.java:834)
2023-09-06 17:43:21 Caused by: org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: []
2023-09-06 17:43:21 [2023-09-06 16:43:21,767] ERROR [oracle-db-source|task-0] WorkerSourceTask{id=oracle-db-source-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:188)