[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/delay_event'
 require "uri"
 require "securerandom"
 require "json"
@@ -43,7 +44,7 @@ class LogStash::Outputs::Doris < LogStash::Outputs::Base
    config :db, :validate => :string, :required => true
    # the table which data is loaded to
    config :table, :validate => :string, :required => true
-   # label prefix of a stream load requst.
+   # label prefix of a stream load request.
    config :label_prefix, :validate => :string, :default => "logstash"
    # user name
    config :user, :validate => :string, :required => true
@@ -72,6 +73,7 @@ class LogStash::Outputs::Doris < LogStash::Outputs::Base
 
    config :log_progress_interval, :validate => :number, :default => 10
 
+   config :retry_queue_size, :validate => :number, :default => 128
 
    def print_plugin_info()
       @plugins = Gem::Specification.find_all{|spec| spec.name =~ /logstash-output-doris/ }
@@ -131,22 +133,36 @@ def register
          end
       end
 
+      @retry_queue = java.util.concurrent.DelayQueue.new
+      retry_thread = Thread.new do
+         while popped = @retry_queue.take
+            documents, http_headers, event_num, req_count = popped.event
+            handle_request(documents, http_headers, event_num, req_count)
+         end
+      end
+
       print_plugin_info()
    end # def register
 
+   private
+   def add_event_to_retry_queue(delay_event, block)
+      if block
+         while @retry_queue.size >= @retry_queue_size
+            sleep(1)
+         end
+      end
+      @retry_queue.add(delay_event)
+   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 +173,59 @@ 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
+
+   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 is not a valid JSON:\n#{response}")
+      end
+
+      status = response_json["Status"]
 
-         status = response_json["Status"]
+      if status == 'Label Already Exists'
+         @logger.warn("Label already exists: #{response_json['Label']}, skip #{event_num} records:\n#{response}")
+         return
+      end
 
-         if status == 'Label Already Exists'
-           @logger.warn("Label already exists: #{response_json['Label']}, skip #{event_num} records.")
-           break
+      if status == "Success" || status == "Publish Timeout"
+         @total_bytes.addAndGet(documents.size)
+         @total_rows.addAndGet(event_num)
+         if @log_request or @logger.debug?
+            @logger.info("doris stream load response:\n#{response}")
          end
+         return
+      end
 
-         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)
+      @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-1 > @max_retries)
+      # if @max_retries >= 0 && req_count-1 > @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 the #{req_count-1}th retry after #{sleep_for} secs.")
+      delay_event = DelayEvent.new(sleep_for, [documents, http_headers, event_num, req_count])
+      add_event_to_retry_queue(delay_event, req_count <= 1)
    end
 
    private
@@ -227,11 +252,7 @@ def make_request(documents, http_headers, query, host)
          log_failure("doris stream load request error: #{e}")
       end
 
-      if @log_request or @logger.debug?
-         @logger.info("doris stream load response:\n#{response}")
-      end
-
-      return response
+      response
    end # def make_request
 
    # Format the HTTP body
@@ -284,8 +305,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/delay_event.rb

@@ -0,0 +1,44 @@
+# 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 DelayEvent
+   include java.util.concurrent.Delayed
+
+   def initialize(delay, event)
+      @start_time = Time.now.to_i + delay
+      @event = event # event style: [documents, http_headers, event_num, req_count]
+   end
+
+   def get_delay(unit)
+      delay = @start_time - Time.now.to_i
+      unit.convert(delay, java.util.concurrent.TimeUnit::SECONDS)
+   end
+
+   def compare_to(other)
+      d = self.get_delay(java.util.concurrent.TimeUnit::SECONDS) - other.get_delay(java.util.concurrent.TimeUnit::SECONDS)
+      if d == 0
+         0
+      end
+      d < 0 ? -1 : 1
+   end
+
+   def event
+      @event
+   end
+end