fix: remove reconnection feature from client library

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java

@@ -65,8 +65,7 @@
  * without offset, please use a simpler writer {@code DirectWriter}.
  *
  * <p>A {@link StreamWrier} provides built-in capabilities to: handle batching of messages;
- * controlling memory utilization (through flow control); automatic connection re-establishment and
- * request cleanup (only keeps write schema on first request in the stream).
+ * controlling memory utilization (through flow control); request cleanup (only keeps write schema on first request in the stream).
  *
  * <p>With customizable options that control:
  *
@@ -496,6 +495,7 @@ private void onFailure(Throwable t) {
         return;
       } else {
         LOG.info("Setting " + t.toString() + " on response");
+        streamWriter.messagesWaiter.release(this.getByteSize());
         this.streamWriter.setException(t);
       }
 
@@ -871,7 +871,6 @@ private void abortInflightRequests(Throwable t) {
                   t,
                   GrpcStatusCode.of(Status.Code.ABORTED),
                   true));
-          streamWriter.messagesWaiter.release(inflightBatch.getByteSize());
         }
       }
     }
@@ -943,44 +942,8 @@ public void onError(Throwable t) {
         }
         inflightBatch = this.inflightBatches.poll();
       }
-      streamWriter.messagesWaiter.release(inflightBatch.getByteSize());
-      if (isRecoverableError(t)) {
-        try {
-          if (streamWriter.currentRetries < streamWriter.getRetrySettings().getMaxAttempts()
-              && !streamWriter.shutdown.get()) {
-            synchronized (streamWriter.currentRetries) {
-              streamWriter.currentRetries++;
-            }
-            LOG.info(
-                "Try to reestablish connection due to transient error: "
-                    + t.toString()
-                    + " retry times: "
-                    + streamWriter.currentRetries);
-            streamWriter.refreshAppend();
-            LOG.info("Resending requests on after connection established");
-            streamWriter.writeBatch(inflightBatch);
-          } else {
-            inflightBatch.onFailure(t);
-            abortInflightRequests(t);
-            synchronized (streamWriter.currentRetries) {
-              streamWriter.currentRetries = 0;
-            }
-          }
-        } catch (InterruptedException e) {
-          LOG.info("Got exception while retrying: " + e.toString());
-          inflightBatch.onFailure(new StatusRuntimeException(Status.ABORTED));
-          abortInflightRequests(new StatusRuntimeException(Status.ABORTED));
-          synchronized (streamWriter.currentRetries) {
-            streamWriter.currentRetries = 0;
-          }
-        }
-      } else {
-        inflightBatch.onFailure(t);
-        abortInflightRequests(t);
-        synchronized (streamWriter.currentRetries) {
-          streamWriter.currentRetries = 0;
-        }
-      }
+      inflightBatch.onFailure(t);
+      abortInflightRequests(t);
     }
   };
 

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java

@@ -32,7 +32,6 @@
 import com.google.api.gax.grpc.testing.LocalChannelProvider;
 import com.google.api.gax.grpc.testing.MockGrpcService;
 import com.google.api.gax.grpc.testing.MockServiceHelper;
-import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.DataLossException;
 import com.google.cloud.bigquery.storage.test.Test.FooType;
 import com.google.common.base.Strings;
@@ -510,42 +509,6 @@ public void testFlowControlBehaviorException() throws Exception {
     }
   }
 
-  @Test
-  public void testStreamReconnectionTransient() throws Exception {
-    StreamWriter writer =
-        getTestStreamWriterBuilder()
-            .setBatchingSettings(
-                StreamWriter.Builder.DEFAULT_BATCHING_SETTINGS
-                    .toBuilder()
-                    .setDelayThreshold(Duration.ofSeconds(100000))
-                    .setElementCountThreshold(1L)
-                    .setFlowControlSettings(
-                        StreamWriter.Builder.DEFAULT_FLOW_CONTROL_SETTINGS
-                            .toBuilder()
-                            .setMaxOutstandingElementCount(1L)
-                            .setLimitExceededBehavior(FlowController.LimitExceededBehavior.Block)
-                            .build())
-                    .build())
-            .build();
-
-    testBigQueryWrite.addResponse(
-        AppendRowsResponse.newBuilder()
-            .setAppendResult(
-                AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build())
-            .build());
-    testBigQueryWrite.addException(new StatusRuntimeException(Status.UNAVAILABLE));
-    testBigQueryWrite.addResponse(
-        AppendRowsResponse.newBuilder()
-            .setAppendResult(
-                AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(1)).build())
-            .build());
-    ApiFuture<AppendRowsResponse> future1 = sendTestMessage(writer, new String[] {"m1"});
-    ApiFuture<AppendRowsResponse> future2 = sendTestMessage(writer, new String[] {"m1"});
-    assertEquals(0L, future1.get().getAppendResult().getOffset().getValue());
-    assertEquals(1L, future2.get().getAppendResult().getOffset().getValue());
-    writer.close();
-  }
-
   @Test
   public void testStreamReconnectionPermanant() throws Exception {
     StreamWriter writer =
@@ -569,36 +532,6 @@ public void testStreamReconnectionPermanant() throws Exception {
     writer.close();
   }
 
-  @Test
-  public void testStreamReconnectionExceedRetry() throws Exception {
-    StreamWriter writer =
-        getTestStreamWriterBuilder()
-            .setBatchingSettings(
-                StreamWriter.Builder.DEFAULT_BATCHING_SETTINGS
-                    .toBuilder()
-                    .setDelayThreshold(Duration.ofSeconds(100000))
-                    .setElementCountThreshold(1L)
-                    .build())
-            .setRetrySettings(
-                RetrySettings.newBuilder()
-                    .setMaxRetryDelay(Duration.ofMillis(100))
-                    .setMaxAttempts(1)
-                    .build())
-            .build();
-    assertEquals(1, writer.getRetrySettings().getMaxAttempts());
-    StatusRuntimeException transientError = new StatusRuntimeException(Status.UNAVAILABLE);
-    testBigQueryWrite.addException(transientError);
-    testBigQueryWrite.addException(transientError);
-    ApiFuture<AppendRowsResponse> future3 = sendTestMessage(writer, new String[] {"toomanyretry"});
-    try {
-      future3.get();
-      Assert.fail("This should fail.");
-    } catch (ExecutionException e) {
-      assertEquals(transientError.toString(), e.getCause().getCause().toString());
-    }
-    writer.close();
-  }
-
   @Test
   public void testOffset() throws Exception {
     try (StreamWriter writer =
@@ -957,43 +890,6 @@ public void testFlushAll() throws Exception {
     writer.close();
   }
 
-  @Test
-  public void testFlushAllFailed() throws Exception {
-    StreamWriter writer =
-        getTestStreamWriterBuilder()
-            .setBatchingSettings(
-                StreamWriter.Builder.DEFAULT_BATCHING_SETTINGS
-                    .toBuilder()
-                    .setElementCountThreshold(2L)
-                    .setDelayThreshold(Duration.ofSeconds(100000))
-                    .build())
-            .build();
-
-    testBigQueryWrite.addException(Status.DATA_LOSS.asException());
-    testBigQueryWrite.addException(Status.DATA_LOSS.asException());
-    testBigQueryWrite.addException(Status.DATA_LOSS.asException());
-
-    ApiFuture<AppendRowsResponse> appendFuture1 = sendTestMessage(writer, new String[] {"A"});
-    ApiFuture<AppendRowsResponse> appendFuture2 = sendTestMessage(writer, new String[] {"B"});
-    ApiFuture<AppendRowsResponse> appendFuture3 = sendTestMessage(writer, new String[] {"C"});
-
-    assertFalse(appendFuture3.isDone());
-    try {
-      writer.flushAll(100000);
-      fail("Should have thrown an Exception");
-    } catch (Exception expected) {
-      if (expected.getCause() instanceof com.google.api.gax.rpc.DataLossException) {
-        LOG.info("got: " + expected.toString());
-      } else {
-        fail("Unexpected exception:" + expected.toString());
-      }
-    }
-
-    assertTrue(appendFuture3.isDone());
-
-    writer.close();
-  }
-
   @Test
   public void testDatasetTraceId() throws Exception {
     StreamWriter writer =