out_rdkafka2: Skip events with an invalid partition or message_key - #576
Merged
Merged
Conversation
Watson1978
force-pushed
the
rdkafka2-invalid-partition
branch
2 times, most recently
from
August 31, 2026 03:54
dd79473 to
fdb069a
Compare
Watson1978
marked this pull request as draft
August 31, 2026 04:34
Watson1978
force-pushed
the
rdkafka2-invalid-partition
branch
from
August 31, 2026 04:39
fdb069a to
1bfd8e7
Compare
Watson1978
marked this pull request as ready for review
August 31, 2026 04:43
kenhys
requested changes
Aug 31, 2026
producer.produce is called outside the per-event rescue, so a record
whose partition is not a valid int32 makes the FFI binding raise
TypeError or RangeError. Neither responds to #code, so
enqueue_with_retry and the outer handler both re-raise it and Fluentd
retries the same chunk forever, stalling every source that feeds this
output.
- Coerce partition inside the per-event guard so a bad value skips only
that event: Integer(value, 10) for strings, and a -1..2**31 - 1 check
- Coerce message_key with to_s, which librdkafka needs for #bytesize
- Put the partition check in KafkaPluginUtil, since out_kafka2 needs the
same one: ruby-kafka does not validate an explicit partition either,
and pack("l>") silently wraps an out of range value
Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Watson1978
force-pushed
the
rdkafka2-invalid-partition
branch
from
August 31, 2026 05:45
1bfd8e7 to
3eeb8d2
Compare
kenhys
approved these changes
Aug 31, 2026
Contributor
Author
|
Thanks |
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
2 participants
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.
producer.produceis called fromenqueue_with_retry, which sits outside the per-eventrescue StandardError ... nextguard inwrite. A record whosepartitionis not a valid int32 makes the rdkafka FFI binding raiseTypeErrororRangeError, and neither responds to#code, so both handlers re-raise and Fluentd retries the same chunk forever. One malformed record stalls the output and fills the buffer, dropping events from every source that feeds it.partitioninside the per-event guard so a bad value skips only that event:Integer(value, 10)for strings, and a-1..2**31 - 1range check. Base 10 is explicit becauseInteger("010")would otherwise be 8, whiledefault_partition 010is 10.message_keywithto_s, which librdkafka needs for#bytesize.KafkaPluginUtilbecauseout_kafka2needs the same one: ruby-kafka does not validate an explicit partition either, andpack("l>")silently wraps an out of range value. That fix is a separate PR.