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

joker-star-l · 2024-12-04T12:08:06Z

What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

The retry task enters the retry queue separately and no longer blocks the normal queue;
Data quality issues are not retried.

Release note

None

Check List (For Author)

Test
- Regression test
- Unit Test
- Manual test (add detailed scripts or steps below)
- No need to test or manual test. Explain why:
  - This is a refactor/code format and no logic has been changed.
  - Previous test can cover this change.
  - No code files have been changed.
  - Other reason
Behavior changed:
- No.
- Yes.
Does this need documentation?
- No.
- Yes.

Check List (For Reviewer who merge this PR)

Confirm the release note
Confirm test cases
Confirm document
Add branch pick label

doris-robot · 2024-12-04T12:08:12Z

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

What problem was fixed (it's best to include specific error reporting information). How it was fixed.
Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
What features were added. Why was this function added?
Which code was refactored and why was this part of the code refactored?
Which functions were optimized and what is the difference before and after the optimization?

joker-star-l · 2024-12-05T17:25:34Z

run buildall

qidaye

LGTM

github-actions · 2024-12-06T04:08:00Z

PR approved by at least one committer and no changes requested.

github-actions · 2024-12-06T04:08:02Z

PR approved by anyone and no changes requested.

xiaokang · 2024-12-09T00:16:08Z

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

+   end
+
+   def run
+      @retry_queue << @event


Is there any limit on length of retry_queue?

Why not just add event to retry_queue but use a task and timer to do it?

No. Because there won't be too many failures.

Because we need to use a timer to make the event wait for a certain amount of time before being queued and retried, rather than retrying immediately.

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

joker-star-l · 2024-12-10T18:30:11Z

run buildall

joker-star-l · 2024-12-17T02:34:55Z

run buildall

joker-star-l · 2024-12-17T09:41:52Z

run buildall

joker-star-l · 2024-12-23T13:06:14Z

run buildall

github-actions · 2024-12-24T07:47:58Z

PR approved by at least one committer and no changes requested.

xiaokang · 2024-12-26T02:37:40Z

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

+      # 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)


I think you should use LinkedBlockingQueue here since it's very offen to do insert and delete.

xiaokang · 2024-12-26T02:38:38Z

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

-         documents << event_body(event) << "\n"
-         event_num += 1
-      end
+      documents = events.map { |event| event_body(event) }.join("\n")


just refactor?

reduce allocation times

xiaokang · 2024-12-26T02:42:52Z

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

+      begin
+         response_json = JSON.parse(response.body)
+      rescue => _
+         @logger.warn("doris stream load response: #{response} is not a valid JSON")


should return or do something else instead of just go ahead.

xiaokang · 2024-12-26T02:44:53Z

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

+
+      @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)


It's no necessary to do special check for DATA_QUALITY_ERROR, since it's should be handled by max_filter_ration config.

xiaokang · 2024-12-26T02:49:44Z

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

+      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)


Why put 0 to count_block_queue?

joker-star-l · 2024-12-27T17:25:52Z

run buildall

xiaokang · 2024-12-30T00:05:15Z

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

+      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)


Why block just for req_count <= 1 ?

xiaokang · 2024-12-30T00:06:41Z

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

@@ -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


128 may be too large if the request batch size is large, eg 100MB. So you should limit on queued bytes instead of items.

xiaokang · 2024-12-30T00:08:08Z

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

-            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)


DATA_QUALITY_ERROR should be processed by setting max_filter_ratio instead of hard code here.

xiaokang · 2024-12-30T00:08:49Z

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

+      begin
+         response_json = JSON.parse(response.body)
+      rescue => _
+         @logger.warn("doris stream load response is not a valid JSON:\n#{response}")


do more exception handling instead of just log warning.

fix: retry

e26d0bb

qidaye previously approved these changes Dec 6, 2024

View reviewed changes

github-actions bot added the approved Indicates a PR has been approved by one committer. label Dec 6, 2024

github-actions bot added the reviewed label Dec 6, 2024

xiaokang reviewed Dec 9, 2024

View reviewed changes

joker-star-l added 2 commits December 11, 2024 01:46

feat: add retry queue size

6ffacbd

fix: queue

50400bc

joker-star-l dismissed qidaye’s stale review via 50400bc December 10, 2024 18:29

github-actions bot removed the approved Indicates a PR has been approved by one committer. label Dec 10, 2024

fix: retry

26a8726

refactor: avoid multiple allocations

f6b023e

refactor: retry condition

211dae0

qidaye previously approved these changes Dec 24, 2024

View reviewed changes

github-actions bot added the approved Indicates a PR has been approved by one committer. label Dec 24, 2024

xiaokang reviewed Dec 26, 2024

View reviewed changes

reactor: retry and log

8829f24

joker-star-l dismissed qidaye’s stale review via 8829f24 December 27, 2024 17:25

github-actions bot removed the approved Indicates a PR has been approved by one committer. label Dec 27, 2024

xiaokang reviewed Dec 30, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

joker-star-l commented Dec 4, 2024 •

edited

Loading

doris-robot commented Dec 4, 2024

joker-star-l commented Dec 5, 2024

qidaye left a comment

github-actions bot commented Dec 6, 2024

github-actions bot commented Dec 6, 2024

xiaokang Dec 9, 2024

joker-star-l Dec 9, 2024

joker-star-l commented Dec 10, 2024

joker-star-l commented Dec 17, 2024

joker-star-l commented Dec 17, 2024

joker-star-l commented Dec 23, 2024

github-actions bot commented Dec 24, 2024

xiaokang Dec 26, 2024

xiaokang Dec 26, 2024

joker-star-l Dec 27, 2024

xiaokang Dec 26, 2024

xiaokang Dec 26, 2024

xiaokang Dec 26, 2024

joker-star-l commented Dec 27, 2024

xiaokang Dec 30, 2024

xiaokang Dec 30, 2024

xiaokang Dec 30, 2024

xiaokang Dec 30, 2024

		@@ -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

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

Are you sure you want to change the base?

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

Conversation

joker-star-l commented Dec 4, 2024 • edited Loading

What problem does this PR solve?

Release note

Check List (For Author)

Check List (For Reviewer who merge this PR)

doris-robot commented Dec 4, 2024

joker-star-l commented Dec 5, 2024

qidaye left a comment

Choose a reason for hiding this comment

github-actions bot commented Dec 6, 2024

github-actions bot commented Dec 6, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joker-star-l commented Dec 10, 2024

joker-star-l commented Dec 17, 2024

joker-star-l commented Dec 17, 2024

joker-star-l commented Dec 23, 2024

github-actions bot commented Dec 24, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joker-star-l commented Dec 27, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

joker-star-l commented Dec 4, 2024 •

edited

Loading