You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "huxihx (JIRA)" <ji...@apache.org> on 2018/03/02 02:49:00 UTC
[jira] [Commented] (KAFKA-6595) Kafka connect commit offset
incorrectly.
[ https://issues.apache.org/jira/browse/KAFKA-6595?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16383062#comment-16383062 ]
huxihx commented on KAFKA-6595:
-------------------------------
Could we simply cancel the flush if we find another thread has already initiated a flushing?
> Kafka connect commit offset incorrectly.
> ----------------------------------------
>
> Key: KAFKA-6595
> URL: https://issues.apache.org/jira/browse/KAFKA-6595
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
> Affects Versions: 0.10.2.0
> Reporter: Hanlin Liu
> Priority: Major
>
> Version: ConfluentPlatform Kafka 3.2.0
> SourceTaskOffsetComitter calls commitOffset() and waits for all incomplete records to be sent. While the task is stopped, commitOffset() is called again by the final block in WorkerSourceTask.execute(), it will throw {{Invalid call to OffsetStorageWriter flush() while already flushing, the framework should not allow this}} exception. This will trigger closing Producer without waiting the flush timeout.
> After 30 seconds, all incomplete records has been forcefully aborted. If the {{offset.flush.timeout.ms}} is configured larger than 30 seconds, WorkerSourceTask will consider those aborted records as sent within flush timeout, which results in incorrectly flushing the source offset.
>
> {code:java}
> // code placeholder
> 2018-02-27 02:59:33,134 INFO [] Stopping connector dp-sqlserver-connector-dptask_455 [pool-3-thread-6][Worker.java:254]
> 2018-02-27 02:59:33,134 INFO [] Stopped connector dp-sqlserver-connector-dptask_455 [pool-3-thread-6][Worker.java:264]
> 2018-02-27 02:59:34,121 ERROR [] Invalid call to OffsetStorageWriter flush() while already flushing, the framework should not allow this [pool-1-thread-13][OffsetStorageWriter.java:110]
> 2018-02-27 02:59:34,121 ERROR [] Task dp-sqlserver-connector-dptask_455-0 threw an uncaught and unrecoverable exception [pool-1-thread-13][WorkerTask.java:141]
> org.apache.kafka.connect.errors.ConnectException: OffsetStorageWriter is already flushing
> at org.apache.kafka.connect.storage.OffsetStorageWriter.beginFlush(OffsetStorageWriter.java:112)
> at org.apache.kafka.connect.runtime.WorkerSourceTask.commitOffsets(WorkerSourceTask.java:294)
> at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:177)
> 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:1142)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> 2018-02-27 03:00:00,734 ERROR [] Graceful stop of task dp-sqlserver-connector-dptask_455-0 failed. [pool-3-thread-1][Worker.java:405]
> 2018-02-27 03:00:04,126 INFO [] Proceeding to force close the producer since pending requests could not be completed within timeout 30 ms. [pool-1-thread-13][KafkaProducer.java:713]
> 2018-02-27 03:00:04,127 ERROR [] dp-sqlserver-connector-dptask_455-0 failed to send record to dptask_455.JF_TEST_11.jf_test_tab_8: {} [kafka-producer-network-thread | producer-31][WorkerSourceTask.java:228]
> java.lang.IllegalStateException: Producer is closed forcefully.
> at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:522)
> at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:502)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:147)
> at java.lang.Thread.run(Thread.java:745)
> 2018-02-27 03:00:09,920 INFO [] Finished WorkerSourceTask{id=dp-sqlserver-connector-dptask_455-0} commitOffsets successfully in 47088 ms [pool-4-thread-1][WorkerSourceTask.java:371]
> {code}
>
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)