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 fd07caf..96ec2ec 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 @@ -4,11 +4,14 @@ import com.google.cloud.bigquery.storage.v1.Exceptions; import com.google.cloud.bigquery.storage.v1.ProtoRows; import com.google.cloud.bigquery.storage.v1.StreamWriter; +import com.google.common.base.Preconditions; import com.vinted.flink.bigquery.metric.AsyncBigQueryStreamMetrics; import com.vinted.flink.bigquery.metric.BigQueryStreamMetrics; import com.vinted.flink.bigquery.model.Rows; import com.vinted.flink.bigquery.sink.ExecutorProvider; import io.grpc.Status; +import io.grpc.StatusException; +import io.grpc.StatusRuntimeException; import org.apache.flink.api.connector.sink2.Sink; import org.apache.flink.connector.base.sink.writer.AsyncSinkWriter; import org.apache.flink.connector.base.sink.writer.BufferedRequestState; @@ -152,7 +155,7 @@ protected void submitRequestEntries(List list, Consumer list, Consumer retry(Throwable t, String traceId, StreamRequest request) { var status = Status.fromThrowable(t); request.setRetries(request.getRetries() - 1);