Add bounded backpressure for producer and consumer event sequences#237
Open
ssikka100 wants to merge 2 commits into
Open
Add bounded backpressure for producer and consumer event sequences#237ssikka100 wants to merge 2 commits into
ssikka100 wants to merge 2 commits into
Conversation
When a topic does not exist or fails authorization, the librdkafka partitioner synchronously fails. If is used, librdkafka allocates an internal headers list and transfers ownership to the message. Upon partitioner failure, librdkafka frees the message (which also frees the headers) and then attempts to free the internal headers pointer again, leading to a double-free crash. This fix switches to , manually creating the headers list and allowing to correctly leave memory lifecycle management to the caller when the partitioner fails. This prevents the double-free crash. It also fixes a pre-existing memory leak where the error pointer from rd_kafka_produceva was not being destroyed.
ssikka100
changed the title
ssikka100
force-pushed
the
fix/producer-backpressure
branch
3 times, most recently
from
e4b1030 to
d8033f5
Compare
Motivation: The events AsyncSequence (used by makeProducerWithEvents and makeConsumerWithEvents) used NIOAsyncSequenceProducer with NoBackPressure strategy. If the user produces messages faster than they consume events, delivery reports accumulate without limit, eventually causing OOM. Go's confluent-kafka-go solves this with a bounded 1M channel that blocks the poller when full. Rust avoids it entirely via per-message oneshots. Modifications: - Switch from NoBackPressure to HighLowWatermark strategy for both producer and consumer event sequences - Add configurable eventsBackpressureLowWatermark and eventsBackpressureHighWatermark to KafkaProducerConfig and KafkaConsumerConfig (defaults: 100k/1M matching Go) - When high watermark is hit, event loop continues polling librdkafka but stops yielding to the events sequence. This ensures sendAndAwait continuations are always serviced (matching Go's per-message channel behavior) while letting librdkafka's internal queue fill naturally until send() returns QueueFull - Consumer always yields during shutdown regardless of backpressure state to ensure rebalance/close signals are delivered - Add isYieldingPaused flag to state machines (producer and consumer) with pauseYielding()/resumeYielding() transitions Result: Producers using send() + events get natural backpressure: buffer fills to 1M events, then librdkafka's C-queue saturates, then send() throws QueueFull. sendAndAwait always works regardless of backpressure state. Graceful shutdown always completes. No more unbounded memory growth.
ssikka100
force-pushed
the
fix/producer-backpressure
branch
from
d8033f5 to
6ed9f72
Compare
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.
Summary
Add watermark-based backpressure to producer and consumer event sequences to prevent unbounded memory growth when events are consumed slower than they are produced.
Problem
The events
AsyncSequence(frommakeProducerWithEvents/makeConsumerWithEvents) usedNoBackPressurestrategy. If users produce faster than they consume events, delivery reports accumulate without limit — potential OOM under sustained load. Go's confluent-kafka-go solves this with a bounded 1M channel that blocks the poller when full.Fix
NoBackPressuretoHighLowWatermarkstrategy (configurable, defaults: 100k low / 1M high matching Go)sendAndAwaitcontinuations are always serviced regardless of backpressure state (matching Go's per-message channel behavior)send()returnsQueueFullrd_kafka_error_t*was not destroyed on produce failure, and switches to saferRD_KAFKA_VTYPE_HEADERSownership pattern (matching Go and Rust)Test plan
producerBackpressureSaturatesCQueue— events not consumed → QueueFullsendAndAwaitWorksWhileBackpressured— continuations serviced during backpressureproducerResumesAfterEventsDrained— send works again after draining eventsgracefulShutdownWhileBackpressured— shutdown completes without hangingproducerBackpressure(integration) — end-to-end with real brokersendAndAwaitWorksDuringBackpressure(integration) — real ack returned during backpressure🤖 Generated with Claude Code