From 0d65f0505157d178850e95249b0b78b52ff9615a Mon Sep 17 00:00:00 2001
From: gintarasm <gintaras.matulas@gmail.com>
Date: Wed, 20 Mar 2024 09:29:10 +0200
Subject: [PATCH] fix: set trace id per append

---
 .../flink/bigquery/sink/async/AsyncBigQuerySinkWriter.java      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/com/vinted/flink/bigquery/sink/async/AsyncBigQuerySinkWriter.java b/src/main/java/com/vinted/flink/bigquery/sink/async/AsyncBigQuerySinkWriter.java
index 96ec2ec..5032e4a 100644
--- a/src/main/java/com/vinted/flink/bigquery/sink/async/AsyncBigQuerySinkWriter.java
+++ b/src/main/java/com/vinted/flink/bigquery/sink/async/AsyncBigQuerySinkWriter.java
@@ -108,8 +108,8 @@ protected final void recreateStreamWriter(String traceId, String streamName, Str
 
     @Override
     protected void submitRequestEntries(List<StreamRequest> list, Consumer<List<StreamRequest>> consumer) {
-        var traceId = UUID.randomUUID().toString();
         var requests = list.stream().map(request -> {
+            var traceId = UUID.randomUUID().toString();
             registerAppendMetrics(request);
             var writer = streamWriter(traceId, request.getStream(), request.getTable());
             logger.trace("Trace-id {}, Writing rows stream {} to steamWriter for {} writer id {}", traceId, request.getStream(), writer.getStreamName(), writer.getWriterId());