From a223581315cb663fd2d3e0f0e68354846c538a91 Mon Sep 17 00:00:00 2001 From: Dipendra Singh Date: Fri, 1 Dec 2023 10:18:24 -0800 Subject: [PATCH] out_rdkafka2: enable idempotency for kafka producers Signed-off-by: Dipendra Singh --- README.md | 1 + lib/fluent/plugin/out_rdkafka2.rb | 2 ++ 2 files changed, 3 insertions(+) diff --git a/README.md b/README.md index e4ad29d..362c10c 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,7 @@ If `ruby-kafka` doesn't fit your kafka environment, check `rdkafka2` plugin inst discard_kafka_delivery_failed (bool) :default => false (No discard) partitioner_hash_function (enum) (crc32|murmur2) :default => 'crc32' share_producer (bool) :default => false + idempotent (bool) :default => false # If you intend to rely on AWS IAM auth to MSK with long lived credentials # https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html diff --git a/lib/fluent/plugin/out_rdkafka2.rb b/lib/fluent/plugin/out_rdkafka2.rb index 0fc18e7..a775c19 100644 --- a/lib/fluent/plugin/out_rdkafka2.rb +++ b/lib/fluent/plugin/out_rdkafka2.rb @@ -71,6 +71,7 @@ class Fluent::Rdkafka2Output < Output config_param :default_message_key, :string, :default => nil config_param :partition_key, :string, :default => 'partition', :desc => "Field for kafka partition" config_param :default_partition, :integer, :default => nil + config_param :idempotent, :bool, :default => false, :desc => 'Enable idempotent producer' config_param :output_data_type, :string, :default => 'json', :obsoleted => "Use section instead" config_param :output_include_tag, :bool, :default => false, :obsoleted => "Use section instead" config_param :output_include_time, :bool, :default => false, :obsoleted => "Use section instead" @@ -313,6 +314,7 @@ def build_config config[:"batch.num.messages"] = @rdkafka_message_max_num if @rdkafka_message_max_num config[:"sasl.username"] = @username if @username config[:"sasl.password"] = @password if @password + config[:"enable.idempotence"] = @idempotent if @idempotent @rdkafka_options.each { |k, v| config[k.to_sym] = v