Put pipeline_id into ThreadContext after entering flush_thread

logstash-plugins · Jan 27, 2021 · ca8c2ce · ca8c2ce
1 parent 5f9242d
commit ca8c2ce
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/lib/logstash/outputs/google_bigquery.rb b/lib/logstash/outputs/google_bigquery.rb
@@ -9,6 +9,8 @@
 require 'fileutils'
 require 'concurrent'
 
+java_import org.apache.logging.log4j.ThreadContext
+
 #
 # === Summary
 #
@@ -275,7 +277,9 @@ def write_to_errors_file(messages, table)
   end
 
   def init_batcher_flush_thread
+    pipeline_id = ThreadContext.get('pipeline.id')
     @flush_thread = Thread.new do
+      ThreadContext.put('pipeline.id', pipeline_id)
       until stopping?
         Stud.stoppable_sleep(@flush_interval_secs) { stopping? }