bugfix

Signed-off-by: Bing Wang <[email protected]>

Author: leo198706

lib/fluent/plugin/kafka_producer_ext.rb

@@ -254,9 +254,8 @@ def assign_partitions!
 
       @pending_message_queue.each do |message|
         partition = message.partition
-
+        partition_count = @cluster.partitions_for(message.topic).count
         begin
-          partition_count = @cluster.partitions_for(message.topic).count
 
           if partition.nil?
             partition = @partitioner.call(partition_count, message)

lib/fluent/plugin/out_kafka2.rb

@@ -403,6 +403,7 @@ def write(chunk)
       rescue Kafka::UnknownTopicOrPartition
         if @use_default_for_unknown_topic && topic != @default_topic
           log.warn "'#{topic}' topic not found. Retry with '#{default_topic}' topic"
+          producer.clear_buffer
           topic = @default_topic
           retry
         end