Conversation
- Multi-Stage Ordered Shutdown Protocol:
- Implemented async Consumer::close().await:
1. Fence Delivery: group.fence_delivery() prevents newly fetched records from reaching the consumer iterator.
2. Stop & Await Fetchers: stop_all() dispatches StopRangeFetch with oneshot reply channels to all range fetcher actors and awaits their exit.
3. Commit Acked Offsets: group.commit() persists only explicitly ack()ed consumer offsets to the cluster.
4. Revoke Ownership: group.clear_effective_ownership() clears local assigned ranges.
5. Leave Group: group.leave() sends SyncConsumerGroupAction::Leave to the coordinator, awaits confirmation, and sets quit = true.
- Drop Idempotency & Defense-in-Depth:
- Updated ConsumerGroup::drop to check quit atomic flag: returns immediately as a zero-cost no-op if close() was called, while retaining a best-effort background task fallback for un-closed drops.
- Added ClientError::ConsumerClosed error variant.
- Added ClientError::Server(ServerError). - Actual terminal server failures now preserve their typed category. - Wrong success variants and ClientResponse::Stop remain UnexpectedResponse. - Existing special handling remains intact for topic existence, stale ranges, producer rejection, and consumer-generation fencing. - Redirect and retry behavior is unchanged.
- Full batches are now producer-owned tasks, so canceling the triggering send() cannot cancel other records. - flush() and close() still wait through flush_gate. - Added deterministic e2e coverage for the cancellation case. - Documented precisely when cancellation can still allow publication.
…t_surviving_senders
- Producer flush()/close() drains are now producer-owned tasks, so canceling the waiting future does not cancel the drain.
- The send gate now covers acceptance/buffer insertion, not acknowledgement waiting. - Threshold batches reserve the flush gate before releasing acceptance, preventing flush() from overtaking detached publication.
Migorithm
added
bug
- Removed accidental Producer implementations of Deref, DerefMut, and From<Arc<Inner>>. - Made producer implementation state private. - Replaced the stale synchronization documentation with an accurate internal comment. - Kept Producer cloning and all intended methods unchanged.
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
closes #188
Summary of Changes
This PR enhances the client SDK (
ProducerandConsumer) with deterministic graceful shutdown protocols, lightweight dual-gate synchronization, cancellation resilience for shared batches, and CLIconfiguration validation.
1. Producer Architecture & Synchronization
send_gate: RwLock<()>: Gates active applicationsend()calls againstflush()andclose().flush_gate: Arc<RwLock<()>>: Gates batch network publication against buffer draining, closing the linger-timer race window.RwLockdeadlock prevention rationale inInnerstruct rustdoc.spawn_flush):tokio::spawnbackground tasks underflush_gate.read().send()future discards only that caller's ACK (oneshot::Receiver), allowing shared batch publication to complete durably for surviving senders.flush()andclose()execute producer-owned background drains that continue to completion even if the caller drops theflush()/close()future.2. Consumer Graceful Shutdown (
Consumer::close)group.fence_delivery()prevents new records from reaching caller iterators.stop_all()dispatchesStopRangeFetchwithoneshotreplies to allRangeFetchActors and awaits their exit.group.commit()persists only explicitlyack()ed consumer offsets.group.clear_effective_ownership()clears local partition assignments.group.leave()sendsSyncConsumerGroupAction::Leaveto the coordinator, awaits confirmation, and setsquit = true.consumer_rxupon shutdown to guarantee immediate termination visibility across all consumer clones.ConsumerGroup::dropchecks thequitatomic flag to makedrop()a no-op whenclose().awaitwas called, preserving a best-effort background leave fallback for un-closed drops.3. Configuration & Error Handling
clapvalue_parser = parse_nonzero_usizeformax_batch_bytesandmax_batch_recordsinBufferConfig, while maintaining runtimeBufferConfig::validate()domain error validation.ClientError::ConsumerClosedand enhanced client redirect error mapping.