out_kafka2: Skip events with an invalid partition - #577
Merged
Conversation
producer.produce is called outside the per-event rescue, and ruby-kafka
coerces the partition with Integer() inside it, so a record carrying a
non numeric partition raises ArgumentError or TypeError there. The outer
handler re-raises and Fluentd retries the same chunk forever, stalling
every source that feeds this output.
- Coerce partition with KafkaPluginUtil::PartitionSettings inside the
per-event guard, so a bad value skips only that event
- Base 10 is now explicit: ruby-kafka reads "010" as octal 8, while
default_partition 010 is 10
- The int32 range check also stops a large partition from being wrapped
silently by pack("l>") when the request is encoded
Signed-off-by: Shizuo Fujita <fujita@clear-code.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
kenhys
approved these changes
Sep 1, 2026
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 outside the per-eventrescue StandardError ... nextguard inwrite, and ruby-kafka coerces the partition withInteger()insideproduce, so a record carrying a non numericpartitionraises there. The outer handler re-raises and Fluentd retries the same chunk forever: the output stalls, the buffer fills, and events from every source feeding it are dropped.partitioncomes straight from the record, so one malformed record is enough. This is the ruby-kafka half of #576.partitionwithKafkaPluginUtil::PartitionSettingsinside the per-event guard, so a bad value skips only that event."010"as octal 8 whiledefault_partition 010is 10.-1..2**31 - 1check also matters here:assign_partitions!only fills in a partition when it is nil, and the encoder packs it withpack("l>"), so2**40is silently wrapped to 0 instead of failing.A partition that is a valid int32 but does not exist on the topic still fails at
deliver_messages, which is outside the guard. Telling deterministic delivery failures apart from transient ones is a larger change and is not included.