KAFKA-17871: avoid blocking the herder thread when producer flushing hangs #18142
Conversation
|
Hi @davide-armand |
davide-armand
force-pushed
the
KAFKA-17871-avoid-blocking-herder-thread
branch
from
1563874 to
fd1bddc
Compare
|
The CI is now green. |
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
|
This PR is being marked as stale since it has not had any activity in 90 days. If you If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact). If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed. |
github-actions
bot
added
stale
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
There was a problem hiding this comment.
Hi @davide-armand Thanks for the fix, and thanks for the test!
Are you able to remove the Thread.sleep calls from the test? These could introduce some flakiness later if run on slow machines. Additionally I think this test leaks threads through the executors. Could you ensure that the executors are closed, and the threads terminate?
…hangs The call to `backingStore.get()` (called by connector task threads through `OffsetStorageReaderImpl.offsets()`) can block for long time waiting for data flush to complete (`KafkaProducer.flush()`). This change moves that call outside the synchronized clause that holds `offsetReadFutures`, so that if `backingStore.get()` hangs then it does not keep `offsetReadFutures` locked. The access to `closed` flag (`closed.get()`) is kept inside the synchronize clause to avoid race condition with `close()`. This is important because `OffsetStorageReaderImpl.close()` needs to lock `offsetReadFutures` as well in order to cancel the futures. Since the herder thread calls `OffsetStorageReaderImpl.close()` when attempting to stops a task, before this change this was resulting in the herder thread hanging indefinetely waiting for `backingStore.get()` to complete.
davide-armand
force-pushed
the
KAFKA-17871-avoid-blocking-herder-thread
branch
from
6d22bc4 to
07f465e
Compare
davide-armand
force-pushed
the
KAFKA-17871-avoid-blocking-herder-thread
branch
from
07f465e to
b90bf7a
Compare
@gharris1727 Pushed a fix in a separate commit, will squash before merging. I cannot get a green build, if I'm not mistaken it fails on unrelated tests. |
The call to
backingStore.get()(called by connector task threads throughOffsetStorageReaderImpl.offsets()) can block for long time waiting for data flush to complete (KafkaProducer.flush()).This change moves that call outside the synchronized clause that holds
offsetReadFutures, so that ifbackingStore.get()hangs then it does not keepoffsetReadFutureslocked. The access toclosedflag (closed.get()) is kept inside the synchronize clause to avoid race condition withclose().This is important because
OffsetStorageReaderImpl.close()needs to lockoffsetReadFuturesas well in order to cancel the futures.Since the herder thread calls
OffsetStorageReaderImpl.close()when attempting to stops a task, before this change this was resulting in the herder thread hanging indefinetely waiting forbackingStore.get()to complete.Committer Checklist (excluded from commit message)