chore: Address review comments

vinted · Dec 21, 2023 · c88cbcf · c88cbcf
1 parent ca3b1c9
commit c88cbcf
Showing 1 changed file with 26 additions and 24 deletions.
diff --git a/src/main/java/com/vinted/flink/bigquery/sink/defaultStream/BigQueryDefaultSinkWriter.java b/src/main/java/com/vinted/flink/bigquery/sink/defaultStream/BigQueryDefaultSinkWriter.java
@@ -105,28 +105,6 @@ public void onSuccess(AppendRowsResponse result) {
             this.parent.inflightRequestCount.arriveAndDeregister();
         }
 
-        private void retryWrite(Throwable t, int newRetryCount) {
-            var status = Status.fromThrowable(t);
-            try {
-                if (newRetryCount > 0) {
-                    logger.warn("Trace-id {} Recoverable error {}. Retrying...", traceId, status.getCode(), t);
-                    this.parent.writeWithRetry(traceId, rows, newRetryCount);
-                } else {
-                    logger.error("Trace-id {} Recoverable error {}. No more retries left", traceId, status.getCode(), t);
-                    this.parent.appendAsyncException = new AppendException(traceId, rows, newRetryCount, t);
-                }
-            } catch (Throwable e) {
-                this.parent.appendAsyncException = new AppendException(traceId, rows, newRetryCount, e);
-            }
-        }
-
-        private void doPauseBeforeRetry() {
-            try {
-                Thread.sleep(parent.clientProvider.writeSettings().getRetryPause().toMillis());
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-        }
 
         @Override
         public void onFailure(Throwable t) {
@@ -146,8 +124,9 @@ public void onFailure(Throwable t) {
                     if (t.getMessage().contains("INVALID_ARGUMENT: MessageSize is too large.")) {
                         Optional.ofNullable(this.parent.metrics.get(rows.getStream())).ifPresent(BigQueryStreamMetrics::incSplitCount);
                         logger.warn("Trace-id {} MessageSize is too large. Splitting batch", traceId);
-                        var first = rows.getData().subList(0, rows.getData().size() / 2);
-                        var second = rows.getData().subList(rows.getData().size() / 2, rows.getData().size());
+                        var data = rows.getData();
+                        var first = data.subList(0, data.size() / 2);
+                        var second = data.subList(data.size() / 2, data.size());
                         try {
                             this.parent.writeWithRetry(traceId, rows.updateBatch(first, rows.getOffset()), retryCount - 1);
                             this.parent.writeWithRetry(traceId, rows.updateBatch(second, rows.getOffset() + first.size()), retryCount - 1);
@@ -163,5 +142,28 @@ public void onFailure(Throwable t) {
             }
             this.parent.inflightRequestCount.arriveAndDeregister();
         }
+
+        private void retryWrite(Throwable t, int newRetryCount) {
+            var status = Status.fromThrowable(t);
+            try {
+                if (newRetryCount > 0) {
+                    logger.warn("Trace-id {} Recoverable error {}. Retrying...", traceId, status.getCode(), t);
+                    this.parent.writeWithRetry(traceId, rows, newRetryCount);
+                } else {
+                    logger.error("Trace-id {} Recoverable error {}. No more retries left", traceId, status.getCode(), t);
+                    this.parent.appendAsyncException = new AppendException(traceId, rows, newRetryCount, t);
+                }
+            } catch (Throwable e) {
+                this.parent.appendAsyncException = new AppendException(traceId, rows, newRetryCount, e);
+            }
+        }
+
+        private void doPauseBeforeRetry() {
+            try {
+                Thread.sleep(parent.clientProvider.writeSettings().getRetryPause().toMillis());
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
     }
 }