From 626618b3eaf81716c1e4d708ed92d20863c33ebf Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Mon, 31 Aug 2026 15:05:35 +0900 Subject: [PATCH] out_kafka2: Skip events with an invalid partition 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 Co-authored-by: Claude Opus 5 --- lib/fluent/plugin/out_kafka2.rb | 2 ++ test/plugin/test_out_kafka2.rb | 52 +++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/lib/fluent/plugin/out_kafka2.rb b/lib/fluent/plugin/out_kafka2.rb index 1e6b377..e07e64b 100644 --- a/lib/fluent/plugin/out_kafka2.rb +++ b/lib/fluent/plugin/out_kafka2.rb @@ -98,6 +98,7 @@ class Fluent::Kafka2Output < Output include Fluent::KafkaPluginUtil::AwsIamSettings include Fluent::KafkaPluginUtil::SSLSettings include Fluent::KafkaPluginUtil::SaslSettings + include Fluent::KafkaPluginUtil::PartitionSettings def initialize super @@ -351,6 +352,7 @@ def write(chunk) partition_key = (@exclude_partition_key ? record.delete(@partition_key_key) : record[@partition_key_key]) || @default_partition_key partition = (@exclude_partition ? record.delete(@partition_key) : record[@partition_key]) || @default_partition message_key = (@exclude_message_key ? record.delete(@message_key_key) : record[@message_key_key]) || @default_message_key + partition = coerce_partition(partition) unless partition.nil? if mutate_headers headers = base_headers.clone diff --git a/test/plugin/test_out_kafka2.rb b/test/plugin/test_out_kafka2.rb index 11736d4..cdf9d04 100644 --- a/test/plugin/test_out_kafka2.rb +++ b/test/plugin/test_out_kafka2.rb @@ -27,6 +27,32 @@ def create_driver(conf = config, tag='test') Fluent::Test::Driver::Output.new(Fluent::Kafka2Output).configure(conf) end + class DummyProducer + attr_reader :produced + + # Coerces like ruby-kafka's Producer#produce, which is where an invalid + # partition raises today + def produce(value, key:, partition_key:, partition:, headers:, create_time:, topic:) + @produced ||= [] + @produced << {value: value && value.to_s, key: key && key.to_s, + partition: partition && Integer(partition)} + end + + def deliver_messages + end + + def clear_buffer + end + end + + def create_chunk(records, tag: 'test') + metadata = Fluent::Plugin::Buffer::Metadata.new(nil, tag, nil) + chunk = Fluent::Plugin::Buffer::MemoryChunk.new(metadata) + chunk.extend(Fluent::ChunkMessagePackEventStreamer) + chunk.append(records.map { |record| [event_time, record].to_msgpack }) + chunk + end + def test_configure assert_nothing_raised(Fluent::ConfigError) { create_driver(base_config) @@ -92,6 +118,32 @@ def test_configure_unsupported_scram_mechanism } end + data("non numeric partition" => "not-a-number", + "hash partition" => {"x" => 1}, + "negative partition" => -2, + "out of int32 range" => 2**31) + def test_write_skips_event_with_invalid_partition(partition) + d = create_driver + producer = DummyProducer.new + stub(d.instance).get_producer { producer } + + assert_nothing_raised { + d.instance.write(create_chunk([{"a" => "b"}, {"a" => "c", "partition" => partition}, {"a" => "d"}])) + } + + assert_equal ['{"a":"b"}', '{"a":"d"}'], producer.produced.collect { |message| message[:value] } + end + + def test_write_keeps_event_with_zero_padded_partition + d = create_driver + producer = DummyProducer.new + stub(d.instance).get_producer { producer } + + d.instance.write(create_chunk([{"a" => "b", "partition" => "010"}])) + + assert_equal [10], producer.produced.collect { |message| message[:partition] } + end + data("crc32" => "crc32", "murmur2" => "murmur2") def test_partitioner_hash_function(data)