[GitHub] [kafka] viktorsomogyi commented on pull request #12671: KAFKA-14250: MirrorSourceTask exception causes the task to fail

[GitBox <gi...@apache.org>]

viktorsomogyi commented on PR #12671:
URL: https://github.com/apache/kafka/pull/12671#issuecomment-1259701968

   @C0urante this problem arises when `stop` is called really fast, right after calling `start` when that is still running. I've attached a log sample to the case but the exception we're getting is this:
   ```
   2022-08-22 17:54:46,544 ERROR org.apache.kafka.connect.runtime.WorkerTask: WorkerSourceTask{id=MirrorSourceConnector-0} Task threw an uncaught and unrecoverable exception
   org.apache.kafka.connect.errors.ConnectException: Failed to fetch offsets.
   	at org.apache.kafka.connect.storage.OffsetStorageReaderImpl.offsets(OffsetStorageReaderImpl.java:114)
   	at org.apache.kafka.connect.storage.OffsetStorageReaderImpl.offset(OffsetStorageReaderImpl.java:63)
   	at org.apache.kafka.connect.mirror.MirrorSourceTask.loadOffset(MirrorSourceTask.java:234)
   	at org.apache.kafka.connect.mirror.MirrorSourceTask.lambda$loadOffsets$4(MirrorSourceTask.java:229)
   	at java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1321)
   	at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
   	at java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1580)
   	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
   	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
   	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
   	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
   	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566)
   	at org.apache.kafka.connect.mirror.MirrorSourceTask.loadOffsets(MirrorSourceTask.java:229)
   	at org.apache.kafka.connect.mirror.MirrorSourceTask.start(MirrorSourceTask.java:99)
   	at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:213)
   	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
   	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
   	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:750)
   Caused by: org.apache.kafka.connect.errors.ConnectException: Offset reader closed while attempting to read offsets. This is likely because the task was been scheduled to stop but has taken longer than the graceful shutdown period to do so.
   	at org.apache.kafka.connect.storage.OffsetStorageReaderImpl.offsets(OffsetStorageReaderImpl.java:103)
   	... 21 common frames omitted
   ```
   In this case the task was getting stopped too fast due to a rebalance I believe when it was still starting up, hence we tried to solve this by propagating it the way shown it the PR and handle it as a CancellationException to allow the task to being able to be cancelled and restarted gracefully.
   In this case I'd expect it to be restarted automatically since the problem is caused by the threading design of the herder as I understand.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org