fix(sdk): error callback should await shard flush on producer shutdown - #3953
fix(sdk): error callback should await shard flush on producer shutdown#3953haubur wants to merge 2 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3953 +/- ##
=============================================
- Coverage 83.99% 69.95% -14.05%
Complexity 1358 1358
=============================================
Files 1215 1215
Lines 170548 146054 -24494
Branches 138292 113799 -24493
=============================================
- Hits 143252 102168 -41084
- Misses 23439 39997 +16558
- Partials 3857 3889 +32
🚀 New features to boost your workflow:
|
| let (stop_tx, _) = broadcast::channel::<()>(1); | ||
|
|
||
| let mut stop_rx = stop_tx.subscribe(); | ||
| // A task that receives errors from shards when writing messages failed. |
There was a problem hiding this comment.
restates what the spawn does, drop it.
|
|
||
| // After shards are closed await the error callback task, | ||
| // that might drain queued errors from the final flush. | ||
| if let Err(e) = self._join_handle.await { |
There was a problem hiding this comment.
_join_handle is awaited here, the underscore prefix says unused.
|
|
||
| // Passes, if the error callback is hit once after final | ||
| // flush from shutdown. | ||
| assert_eq!(error_called.load(Ordering::SeqCst), 1); |
There was a problem hiding this comment.
the stub drops ctx, so this proves the callback ran but not that the failed batch came back with it - which is the point of #3947. record ctx.messages.len() in the stub and assert it is 1.
| let mut mock = MockProducerCoreBackend::new(); | ||
| // mock a failed send to be caught by the error callback | ||
| mock.expect_send_internal() | ||
| .times(1) |
There was a problem hiding this comment.
.times(1) is not enforced here - the last Arc owner of the mock is the shard task, so a violated expectation panics there and shutdown() swallows it as "shard panicked". the error_called assert carries the test, just don't rely on times.
| #[tokio::test] | ||
| async fn test_shutdown_reports_errors_from_final_flush() { | ||
| let mut mock = MockProducerCoreBackend::new(); | ||
| // mock a failed send to be caught by the error callback |
There was a problem hiding this comment.
also at 448, 466, 469. these restate the code, drop them (and the stray "block/ ").
github-actions
Bot
added
S-waiting-on-author
2 participants
Which issue does this PR address?
Closes #3947
Rationale
On shutdown() the IggyProducer calls the shutdown() method of the Dispatcher which broadcasts
a termination signal to it's workers. This immediately breaks the loop of the error callback task, which then
does not receive any errors from the final flush anymore.
What changed?
The error callback task does not receive the stop signal anymore. The channel is still open,
and shards can put errors in the queue. On join in shutdown() (after shard tasks are joined) the
error callback is awaited for as long as the queue is emptied.
Local Execution
AI Usage
Test proposed by Claude code.