dekaf: Fix log forwarding dropping some messages #2055
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ll their references in order to prevent inadvertently cancelling a pending append when another message arrives. I realized we were dropping some log messages on the floor, especially messages that were logged soon before a task exited. I ended up realizing that because the append futures were being directly provided to `tokio::select!`, they will get cancelled if another future in the select resolves first. This was happening most noticeably when a Session crashed: the error log event containing the error didn't have enough time to fully append before the `mpsc::Receiver<TaskWriterMessage>` closed, which caused `tokio::select!` to cancel the call to `append_logs_to_writer`, dropping the logs contained within it on the ground.
8 tasks
jshearer
force-pushed
the
dekaf/fix_dropping_logs
branch
from
f0d71b4 to
3a6c574
Compare
We never ended up setting a task name so no logs got sent, and it's best if we don't allow external data like from HTTP requests to get into the ops logs
jshearer
force-pushed
the
dekaf/fix_dropping_logs
branch
from
3a6c574 to
3707c83
Compare
|
Confirmed log forwarding still works locally and in |
psFried
approved these changes
Apr 10, 2025
There was a problem hiding this comment.
I'm good with merging this as is, and I don't see any problems. My general impression is that the start function could still be made simpler and clearer, and that handling logs that were generated before the shard is known doesn't seem worth the complexity. But I'm comfortable merging as is, and either deferring or abandoning further refactoring.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description:
I realized we were dropping some log messages on the floor, especially messages that were logged soon before a task exited. I ended up realizing that because the append futures were being directly provided to
tokio::select!, they will get cancelled if another future in the select resolves first. Per convo w/ @jgraettinger, I refactored the append loop to:MaybeFuture::FutureMaybeFutures if they're notDoneorGoneAlso per convo w/ @psFried, remove log forwarding from the registry. We never ended up setting a task name so no logs got sent, and it's best if we don't allow external data like from HTTP requests to get into the ops logs.
This change is