[enhancement](plugin) logstash: add retry queue without blocking tasks

extension/logstash/lib/logstash/outputs/doris.rb

@@ -22,6 +22,7 @@
 require "logstash/namespace"
 require "logstash/json"
 require 'logstash/util/formater'
+require 'logstash/util/retry_timer_task'
 require "uri"
 require "securerandom"
 require "json"
@@ -131,22 +132,35 @@ def register
          end
       end
 
+      # Run named Timer as daemon thread
+      @timer = java.util.Timer.new("Doris Output #{self.params['id']}", true)
+      # The queue in Timer is unbounded and uncontrollable, so use a new queue to control the amount
+      @count_block_queue = java.util.concurrent.ArrayBlockingQueue.new(128)
+
+      @retry_queue = Queue.new
+      retry_thread = Thread.new do
+         while popped = @retry_queue.pop
+            documents, http_headers, event_num, req_count = popped
+            handle_request(documents, http_headers, event_num, req_count)
+         end
+      end
+
       print_plugin_info()
    end # def register
 
+   def close
+      @timer.cancel
+   end
+
    def multi_receive(events)
       return if events.empty?
       send_events(events)
    end
 
    private
    def send_events(events)
-      documents = ""
-      event_num = 0
-      events.each do |event|
-         documents << event_body(event) << "\n"
-         event_num += 1
-      end
+      documents = events.map { |event| event_body(event) }.join("\n")
+      event_num = events.size
 
       # @logger.info("get event num: #{event_num}")
       @logger.debug("get documents: #{documents}")
@@ -157,50 +171,56 @@ def send_events(events)
          http_headers["label"] = @label_prefix + "_" + @db + "_" + @table + "_" + Time.now.strftime('%Y%m%d_%H%M%S_%L_' + SecureRandom.uuid)
       end
 
-      req_count = 0
-      sleep_for = 1
-      while true
-         response = make_request(documents, http_headers, @http_query, @http_hosts.sample)
-
-         req_count += 1
-         response_json = {}
-         begin
-            response_json = JSON.parse(response.body)
-         rescue => e
-            @logger.warn("doris stream load response: #{response} is not a valid JSON")
-         end
+      handle_request(documents, http_headers, event_num, 1)
+   end
+
+   def sleep_for_attempt(attempt)
+      sleep_for = attempt**2
+      sleep_for = sleep_for <= 60 ? sleep_for : 60
+      (sleep_for/2) + (rand(0..sleep_for)/2)
+   end
 
-         status = response_json["Status"]
+   private
+   def handle_request(documents, http_headers, event_num, req_count)
+      response = make_request(documents, http_headers, @http_query, @http_hosts.sample)
+      response_json = {}
+      begin
+         response_json = JSON.parse(response.body)
+      rescue => _
+         @logger.warn("doris stream load response: #{response} is not a valid JSON")
+      end
 
-         if status == 'Label Already Exists'
-           @logger.warn("Label already exists: #{response_json['Label']}, skip #{event_num} records.")
-           break
-         end
+      status = response_json["Status"]
 
-         if status == "Success" || status == "Publish Timeout"
-            @total_bytes.addAndGet(documents.size)
-            @total_rows.addAndGet(event_num)
-            break
-         else
-            @logger.warn("FAILED doris stream load response:\n#{response}")
-
-            if @max_retries >= 0 && req_count > @max_retries
-               @logger.warn("DROP this batch after failed #{req_count} times.")
-               if @save_on_failure
-                  @logger.warn("Try save to disk.Disk file path : #{@save_dir}/#{@table}_#{@save_file}")
-                  save_to_disk(documents)
-               end
-               break
-            end
-
-            # sleep and then retry
-            sleep_for = sleep_for * 2
-            sleep_for = sleep_for <= 60 ? sleep_for : 60
-            sleep_rand = (sleep_for / 2) + (rand(0..sleep_for) / 2)
-            @logger.warn("Will do retry #{req_count} after sleep #{sleep_rand} secs.")
-            sleep(sleep_rand)
+      if status == 'Label Already Exists'
+         @logger.warn("Label already exists: #{response_json['Label']}, skip #{event_num} records.")
+         return
+      end
+
+      if status == "Success" || status == "Publish Timeout"
+         @total_bytes.addAndGet(documents.size)
+         @total_rows.addAndGet(event_num)
+         return
+      end
+
+      @logger.warn("FAILED doris stream load response:\n#{response}")
+      # if there are data quality issues, we do not retry
+      if (status == 'Fail' && response_json['Message'].start_with?("[DATA_QUALITY_ERROR]")) || (@max_retries >= 0 && req_count > @max_retries)
+         @logger.warn("DROP this batch after failed #{req_count} times.")
+         if @save_on_failure
+            @logger.warn("Try save to disk.Disk file path : #{@save_dir}/#{@table}_#{@save_file}")
+            save_to_disk(documents)
          end
+         return
       end
+
+      # add to retry_queue
+      sleep_for = sleep_for_attempt(req_count)
+      req_count += 1
+      @logger.warn("Will do retry #{req_count} after #{sleep_for} secs.")
+      timer_task = RetryTimerTask.new(@retry_queue, @count_block_queue, [documents, http_headers, event_num, req_count])
+      @count_block_queue.put(0)
+      @timer.schedule(timer_task, sleep_for*1000)
    end
 
    private
@@ -284,8 +304,8 @@ def save_to_disk(documents)
    end
 
     # This is split into a separate method mostly to help testing
-   def log_failure(message)
-      @logger.warn("[Doris Output Failure] #{message}")
+   def log_failure(message, data = {})
+      @logger.warn("[Doris Output Failure] #{message}", data)
    end
 
    def make_request_headers()

extension/logstash/lib/logstash/util/retry_timer_task.rb

@@ -0,0 +1,33 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+require 'java'
+
+class RetryTimerTask < java.util.TimerTask
+   def initialize(retry_queue, count_block_queue, event)
+      @retry_queue = retry_queue
+      @count_block_queue = count_block_queue
+      # event style: [documents, http_headers, event_num, req_count]
+      @event = event
+      super()
+   end
+
+   def run
+      @retry_queue << @event
+      @count_block_queue.poll
+   end
+end