This is an automated email from the ASF dual-hosted git repository.
chamikaramj pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 932dbe68f7f [Java][BigQueryIO] Fix BigQueryIO streaming insert
silently dropping transient network errors during response body read (#40420)
932dbe68f7f is described below
commit 932dbe68f7f7961ca0bb9221815b8f42040e87c0
Author: Danish Amjad <[email protected]>
AuthorDate: Fri Oct 23 02:30:13 2026 +0200
[Java][BigQueryIO] Fix BigQueryIO streaming insert silently dropping
transient network errors during response body read (#40420)
---
.../sdk/io/gcp/bigquery/BigQueryServicesImpl.java | 50 ++++++++
.../io/gcp/bigquery/BigQueryServicesImplTest.java | 141 +++++++++++++++++++++
2 files changed, 191 insertions(+)
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
index a921be3700e..81ba6ce3ca7 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
@@ -99,6 +99,8 @@ import io.grpc.Status;
import io.grpc.Status.Code;
import io.grpc.protobuf.ProtoUtils;
import java.io.IOException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
@@ -1081,6 +1083,27 @@ public class BigQueryServicesImpl implements
BigQueryServices {
} catch (IOException e) {
GoogleJsonError.ErrorInfo errorInfo = getErrorInfo(e);
if (errorInfo == null) {
+ String networkErrorReason = transientNetworkErrorReason(e);
+ if (networkErrorReason != null) {
+ // Route all rows in this batch through InsertRetryPolicy so
the outer insertAll
+ // loop handles retries consistently via the pipeline's
configured policy.
+ serviceCallMetric.call(networkErrorReason);
+ result.updateFailedRpcMetrics(start, start,
BigQuerySinkMetrics.UNKNOWN);
+ LOG.warn(
+ "BigQuery insertAll transient network error, routing rows
to retry policy", e);
+ List<TableDataInsertAllResponse.InsertErrors> syntheticErrors
= new ArrayList<>();
+ for (int i = 0; i < rows.size(); i++) {
+ syntheticErrors.add(
+ new TableDataInsertAllResponse.InsertErrors()
+ .setIndex((long) i)
+ .setErrors(
+ Collections.singletonList(
+ new ErrorProto()
+ .setReason(networkErrorReason)
+
.setMessage(Strings.nullToEmpty(e.getMessage())))));
+ }
+ return syntheticErrors;
+ }
serviceCallMetric.call(ServiceCallMetric.CANONICAL_STATUS_UNKNOWN);
result.updateFailedRpcMetrics(start, start,
BigQuerySinkMetrics.UNKNOWN);
throw e;
@@ -1481,6 +1504,33 @@ public class BigQueryServicesImpl implements
BigQueryServices {
successfulRows);
}
+ /**
+ * Returns a BigQuery-style error reason string for transient network
errors that occur outside
+ * the scope of {@link com.google.api.client.http.HttpRequest#execute}
(e.g. during chunked
+ * response body reading), or {@code null} if the exception is not a
transient network error.
+ */
+ static @Nullable String transientNetworkErrorReason(IOException e) {
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof SocketTimeoutException) {
+ return "socketTimeout";
+ }
+ if (cause instanceof SocketException) {
+ String msg = cause.getMessage();
+ if (msg != null) {
+ if (msg.contains("Connection reset")) {
+ return "connectionReset";
+ }
+ if (msg.contains("Broken pipe")) {
+ return "brokenPipe";
+ }
+ }
+ }
+ cause = cause.getCause();
+ }
+ return null;
+ }
+
protected static GoogleJsonError.@Nullable ErrorInfo
getErrorInfo(IOException e) {
if (!(e instanceof GoogleJsonResponseException)) {
return null;
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
index 642d35b4390..f9ca23fe227 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
@@ -84,6 +84,8 @@ import io.grpc.Status;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -822,6 +824,145 @@ public class BigQueryServicesImplTest {
verifyWriteMetricWasSet("project", "dataset", "table", "quotaexceeded", 1);
}
+ @Test
+ public void testTransientNetworkErrorReason_socketTimeout() {
+ assertEquals(
+ "socketTimeout",
+ DatasetServiceImpl.transientNetworkErrorReason(
+ new SocketTimeoutException("Read timed out")));
+ }
+
+ @Test
+ public void testTransientNetworkErrorReason_wrappedSocketTimeout() {
+ assertEquals(
+ "socketTimeout",
+ DatasetServiceImpl.transientNetworkErrorReason(
+ new IOException("wrapper", new SocketTimeoutException("Read timed
out"))));
+ }
+
+ @Test
+ public void testTransientNetworkErrorReason_connectionReset() {
+ assertEquals(
+ "connectionReset",
+ DatasetServiceImpl.transientNetworkErrorReason(new
SocketException("Connection reset")));
+ }
+
+ @Test
+ public void testTransientNetworkErrorReason_brokenPipe() {
+ assertEquals(
+ "brokenPipe",
+ DatasetServiceImpl.transientNetworkErrorReason(new
SocketException("Broken pipe")));
+ }
+
+ @Test
+ public void testTransientNetworkErrorReason_nonTransient() {
+ assertNull(DatasetServiceImpl.transientNetworkErrorReason(new
IOException("Some other error")));
+ }
+
+ /**
+ * Tests that {@link DatasetServiceImpl#insertAll} retries when a
SocketTimeoutException occurs
+ * during response body reading (outside RetryHttpRequestInitializer scope).
+ */
+ @Test
+ public void testInsertSocketTimeoutRetry() throws Exception {
+ TableReference ref =
+ new
TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
+ List<FailsafeValueInSingleWindow<TableRow, TableRow>> rows = new
ArrayList<>();
+ rows.add(wrapValue(new TableRow()));
+
+ // First response: 200 OK but content stream throws SocketTimeoutException
during body read,
+ // simulating a timeout mid-chunked-response. Second response succeeds.
+ setupMockResponses(
+ response -> {
+ when(response.getStatusCode()).thenReturn(200);
+ when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
+ when(response.getContent())
+ .thenReturn(
+ new InputStream() {
+ @Override
+ public int read() throws IOException {
+ throw new SocketTimeoutException("Read timed out");
+ }
+ });
+ },
+ response -> {
+ when(response.getStatusCode()).thenReturn(200);
+ when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
+ when(response.getContent()).thenReturn(toStream(new
TableDataInsertAllResponse()));
+ });
+
+ DatasetServiceImpl dataService =
+ new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());
+ dataService.insertAll(
+ ref,
+ rows,
+ null,
+ BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()),
+ TEST_BACKOFF,
+ new MockSleeper(),
+ InsertRetryPolicy.alwaysRetry(),
+ null,
+ null,
+ false,
+ false,
+ false,
+ null);
+
+ verifyAllResponsesAreRead();
+ expectedLogs.verifyWarn(
+ "BigQuery insertAll transient network error, routing rows to retry
policy");
+ }
+
+ /**
+ * Tests that {@link DatasetServiceImpl#insertAll} throws IOException when
SocketTimeoutException
+ * retries are exhausted via the outer insertAll backoff.
+ */
+ @Test
+ public void testInsertSocketTimeoutExhausted() throws Exception {
+ TableReference ref =
+ new
TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
+ List<FailsafeValueInSingleWindow<TableRow, TableRow>> rows = new
ArrayList<>();
+ rows.add(wrapValue(new TableRow()));
+
+ // Single response that times out during body read. With maxRetries(0) on
the outer backoff
+ // there is no retry; the synthetic timeout errors are collected and
IOException is thrown.
+ setupMockResponses(
+ response -> {
+ when(response.getStatusCode()).thenReturn(200);
+ when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
+ when(response.getContent())
+ .thenReturn(
+ new InputStream() {
+ @Override
+ public int read() throws IOException {
+ throw new SocketTimeoutException("Read timed out");
+ }
+ });
+ });
+
+ DatasetServiceImpl dataService =
+ new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());
+ // Rows are routed through InsertRetryPolicy as synthetic "timeout"
errors; when the outer
+ // backoff is exhausted insertAll throws IOException("Insert failed: ...").
+ assertThrows(
+ IOException.class,
+ () ->
+ dataService.insertAll(
+ ref,
+ rows,
+ null,
+
BackOffAdapter.toGcpBackOff(FluentBackoff.DEFAULT.withMaxRetries(0).backoff()),
+ FluentBackoff.DEFAULT.withMaxRetries(0),
+ new MockSleeper(),
+ InsertRetryPolicy.alwaysRetry(),
+ null,
+ null,
+ false,
+ false,
+ false,
+ null));
+ }
+
/** Tests that {@link DatasetServiceImpl#insertAll} can stop quotaExceeded
retry attempts. */
@Test
public void testInsertStoppedRetry() throws Exception {
