Merge pull request #466 from raytung/fix/out_kafka2_shared_producer_send_size

fix(out_rdkafka2): fixed a bug where it is not respecting the Buffer sections' `chunk_limit_records` and `chunk_limit_size` options

Author: ashie

lib/fluent/plugin/kafka_producer_ext.rb

@@ -38,9 +38,15 @@ def produce_for_buffered(value, key: nil, topic:, partition: nil, partition_key:
 end
 
 # for out_kafka2
+# Majority (if not all) of this code is lifted from https://github.com/zendesk/ruby-kafka/blob/master/lib/kafka/producer.rb
+# with the main difference where we have removed any checks regarding max_buffer_bytesize and max_buffer_size
+# The reason for doing this is to provide a better UX for our users where they only need to set those bounds in
+# the Buffer section using `chunk_limit_size` and `chunk_limit_records`.
+#
+# We should reconsider this in the future in case the `ruby-kafka` library drastically changes its internal.
 module Kafka
   class Client
-    def topic_producer(topic, compression_codec: nil, compression_threshold: 1, ack_timeout: 5, required_acks: :all, max_retries: 2, retry_backoff: 1, max_buffer_size: 1000, max_buffer_bytesize: 10_000_000, idempotent: false, transactional: false, transactional_id: nil, transactional_timeout: 60)
+    def custom_producer(compression_codec: nil, compression_threshold: 1, ack_timeout: 5, required_acks: :all, max_retries: 2, retry_backoff: 1, max_buffer_size: 1000, max_buffer_bytesize: 10_000_000, idempotent: false, transactional: false, transactional_id: nil, transactional_timeout: 60)
       cluster = initialize_cluster
       compressor = Compressor.new(
         codec_name: compression_codec,
@@ -57,8 +63,7 @@ def topic_producer(topic, compression_codec: nil, compression_threshold: 1, ack_
         transactional_timeout: transactional_timeout,
       )
 
-      TopicProducer.new(topic,
-        cluster: cluster,
+      CustomProducer.new(cluster: cluster,
         transaction_manager: transaction_manager,
         logger: @logger,
         instrumenter: @instrumenter,
@@ -74,8 +79,8 @@ def topic_producer(topic, compression_codec: nil, compression_threshold: 1, ack_
     end
   end
 
-  class TopicProducer
-    def initialize(topic, cluster:, transaction_manager:, logger:, instrumenter:, compressor:, ack_timeout:, required_acks:, max_retries:, retry_backoff:, max_buffer_size:, max_buffer_bytesize:, partitioner:)
+  class CustomProducer
+    def initialize(cluster:, transaction_manager:, logger:, instrumenter:, compressor:, ack_timeout:, required_acks:, max_retries:, retry_backoff:, max_buffer_size:, max_buffer_bytesize:, partitioner:)
       @cluster = cluster
       @transaction_manager = transaction_manager
       @logger = logger
@@ -88,23 +93,19 @@ def initialize(topic, cluster:, transaction_manager:, logger:, instrumenter:, co
       @max_buffer_bytesize = max_buffer_bytesize
       @compressor = compressor
       @partitioner = partitioner
-
-      @topic = topic
-      @cluster.add_target_topics(Set.new([topic]))
-
       # A buffer organized by topic/partition.
       @buffer = MessageBuffer.new
 
       # Messages added by `#produce` but not yet assigned a partition.
       @pending_message_queue = PendingMessageQueue.new
     end
 
-    def produce(value, key: nil, partition: nil, partition_key: nil, headers: EMPTY_HEADER, create_time: Time.now)
+    def produce(value, key: nil, partition: nil, partition_key: nil, headers: EMPTY_HEADER, create_time: Time.now, topic: nil)
       message = PendingMessage.new(
         value: value,
         key: key,
         headers: headers,
-        topic: @topic,
+        topic: topic,
         partition: partition,
         partition_key: partition_key,
         create_time: create_time
@@ -245,12 +246,13 @@ def deliver_messages_with_retries
 
     def assign_partitions!
       failed_messages = []
-      partition_count = @cluster.partitions_for(@topic).count
 
       @pending_message_queue.each do |message|
         partition = message.partition
 
         begin
+          partition_count = @cluster.partitions_for(message.topic).count
+
           if partition.nil?
             partition = @partitioner.call(partition_count, message)
           end

lib/fluent/plugin/out_kafka2.rb

@@ -207,7 +207,7 @@ def multi_workers_ready?
     end
 
     def create_producer
-      @kafka.producer(**@producer_opts)
+      @kafka.custom_producer(**@producer_opts)
     end
 
     def start