From 9475360c712a1d6fd153ac7b0e27f3a7986542b0 Mon Sep 17 00:00:00 2001 From: yirutang Date: Thu, 14 Jan 2021 15:41:50 -0800 Subject: [PATCH 1/6] fix:a hang in StreamWriter when inflight request reached a limit and we try to resend a message --- .../storage/v1beta2/StreamWriter.java | 54 ++++++++++--------- .../storage/v1beta2/StreamWriterTest.java | 19 +++++-- 2 files changed, 43 insertions(+), 30 deletions(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java index 88837e8582..baf9fbc0ff 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java @@ -922,41 +922,43 @@ public void onError(Throwable t) { } inflightBatch = this.inflightBatches.poll(); } - try { - if (isRecoverableError(t)) { - try { - if (streamWriter.currentRetries < streamWriter.getRetrySettings().getMaxAttempts() - && !streamWriter.shutdown.get()) { - streamWriter.refreshAppend(); - LOG.info("Resending requests on transient error:" + streamWriter.currentRetries); - streamWriter.writeBatch(inflightBatch); - synchronized (streamWriter.currentRetries) { - streamWriter.currentRetries++; - } - } else { - inflightBatch.onFailure(t); - abortInflightRequests(t); - synchronized (streamWriter.currentRetries) { - streamWriter.currentRetries = 0; - } + streamWriter.messagesWaiter.release(inflightBatch.getByteSize()); + if (isRecoverableError(t)) { + try { + if (streamWriter.currentRetries < streamWriter.getRetrySettings().getMaxAttempts() + && !streamWriter.shutdown.get()) { + synchronized (streamWriter.currentRetries) { + streamWriter.currentRetries++; } - } catch (IOException | InterruptedException e) { - LOG.info("Got exception while retrying."); - inflightBatch.onFailure(e); - abortInflightRequests(e); + 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; } } - } else { - inflightBatch.onFailure(t); - abortInflightRequests(t); + } catch (IOException | InterruptedException e) { + LOG.info("Got exception while retrying."); + inflightBatch.onFailure(e); + abortInflightRequests(e); synchronized (streamWriter.currentRetries) { streamWriter.currentRetries = 0; } } - } finally { - streamWriter.messagesWaiter.release(inflightBatch.getByteSize()); + } else { + inflightBatch.onFailure(t); + abortInflightRequests(t); + synchronized (streamWriter.currentRetries) { + streamWriter.currentRetries = 0; + } } } }; diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java index ddaee5f34b..1d25662087 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java @@ -519,21 +519,32 @@ public void testStreamReconnectionTransient() throws Exception { .toBuilder() .setDelayThreshold(Duration.ofSeconds(100000)) .setElementCountThreshold(1L) + .setFlowControlSettings( + StreamWriter.Builder.DEFAULT_FLOW_CONTROL_SETTINGS + .toBuilder() + .setMaxOutstandingElementCount(1L) + .setLimitExceededBehavior( + FlowController.LimitExceededBehavior.Block) + .build()) .build()) .build(); StatusRuntimeException transientError = new StatusRuntimeException(Status.UNAVAILABLE); - testBigQueryWrite.addException(transientError); testBigQueryWrite.addResponse( AppendRowsResponse.newBuilder() .setAppendResult( AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build()) .build()); + testBigQueryWrite.addException(transientError); + testBigQueryWrite.addResponse( + AppendRowsResponse.newBuilder() + .setAppendResult( + AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(1)).build()) + .build()); ApiFuture future1 = sendTestMessage(writer, new String[] {"m1"}); - assertEquals(false, future1.isDone()); - // Retry is scheduled to be 7 seconds later. + ApiFuture future2 = sendTestMessage(writer, new String[] {"m1"}); assertEquals(0L, future1.get().getAppendResult().getOffset().getValue()); - future1.get(); + assertEquals(1L, future2.get().getAppendResult().getOffset().getValue()); writer.close(); } From b297eff71176351d47e7afc342cab2ddaa2a5786 Mon Sep 17 00:00:00 2001 From: yirutang Date: Thu, 14 Jan 2021 15:42:06 -0800 Subject: [PATCH 2/6] . --- .../cloud/bigquery/storage/v1beta2/StreamWriterTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java index 1d25662087..d77b8de109 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java @@ -523,8 +523,7 @@ public void testStreamReconnectionTransient() throws Exception { StreamWriter.Builder.DEFAULT_FLOW_CONTROL_SETTINGS .toBuilder() .setMaxOutstandingElementCount(1L) - .setLimitExceededBehavior( - FlowController.LimitExceededBehavior.Block) + .setLimitExceededBehavior(FlowController.LimitExceededBehavior.Block) .build()) .build()) .build(); From 4b35a9934cf246115bde9aabe017830ab5685769 Mon Sep 17 00:00:00 2001 From: yirutang Date: Thu, 14 Jan 2021 15:57:35 -0800 Subject: [PATCH 3/6] . --- .../com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java index baf9fbc0ff..97f0a8c27d 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java @@ -946,7 +946,7 @@ public void onError(Throwable t) { } } } catch (IOException | InterruptedException e) { - LOG.info("Got exception while retrying."); + LOG.info("Got exception while retrying: " + e.toString()); inflightBatch.onFailure(e); abortInflightRequests(e); synchronized (streamWriter.currentRetries) { From b81bca5b850587db6f958a161561b4378e1d2bc5 Mon Sep 17 00:00:00 2001 From: yirutang Date: Fri, 15 Jan 2021 12:13:55 -0800 Subject: [PATCH 4/6] . --- .../storage/v1beta2/StreamWriter.java | 26 +++++++++++-------- .../storage/v1beta2/StreamWriterTest.java | 3 +-- 2 files changed, 16 insertions(+), 13 deletions(-) diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java index 97f0a8c27d..64b11321de 100644 --- a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java +++ b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriter.java @@ -187,7 +187,14 @@ private StreamWriter(Builder builder) this.onSchemaUpdateRunnable.setStreamWriter(this); } - refreshAppend(); + bidiStreamingCallable = stub.appendRowsCallable(); + clientStream = bidiStreamingCallable.splitCall(responseObserver); + try { + while (!clientStream.isSendReady()) { + Thread.sleep(10); + } + } catch (InterruptedException e) { + } } /** Stream name we are writing to. */ @@ -296,9 +303,9 @@ public void flushAll(long timeoutMillis) throws Exception { /** * Re-establishes a stream connection. * - * @throws IOException + * @throws InterruptedException */ - public void refreshAppend() throws IOException, InterruptedException { + public void refreshAppend() throws InterruptedException { appendAndRefreshAppendLock.lock(); if (shutdown.get()) { LOG.warning("Cannot refresh on a already shutdown writer."); @@ -313,11 +320,8 @@ public void refreshAppend() throws IOException, InterruptedException { messagesBatch.resetAttachSchema(); bidiStreamingCallable = stub.appendRowsCallable(); clientStream = bidiStreamingCallable.splitCall(responseObserver); - try { - while (!clientStream.isSendReady()) { - Thread.sleep(10); - } - } catch (InterruptedException expected) { + while (!clientStream.isSendReady()) { + Thread.sleep(10); } Thread.sleep(this.retrySettings.getInitialRetryDelay().toMillis()); // Can only unlock here since need to sleep the full 7 seconds before stream can allow appends. @@ -945,10 +949,10 @@ public void onError(Throwable t) { streamWriter.currentRetries = 0; } } - } catch (IOException | InterruptedException e) { + } catch (InterruptedException e) { LOG.info("Got exception while retrying: " + e.toString()); - inflightBatch.onFailure(e); - abortInflightRequests(e); + inflightBatch.onFailure(new StatusRuntimeException(Status.ABORTED)); + abortInflightRequests(new StatusRuntimeException(Status.ABORTED)); synchronized (streamWriter.currentRetries) { streamWriter.currentRetries = 0; } diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java index d77b8de109..7cc101de19 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1beta2/StreamWriterTest.java @@ -528,13 +528,12 @@ public void testStreamReconnectionTransient() throws Exception { .build()) .build(); - StatusRuntimeException transientError = new StatusRuntimeException(Status.UNAVAILABLE); testBigQueryWrite.addResponse( AppendRowsResponse.newBuilder() .setAppendResult( AppendRowsResponse.AppendResult.newBuilder().setOffset(Int64Value.of(0)).build()) .build()); - testBigQueryWrite.addException(transientError); + testBigQueryWrite.addException(new StatusRuntimeException(Status.UNAVAILABLE)); testBigQueryWrite.addResponse( AppendRowsResponse.newBuilder() .setAppendResult( From 2b22efe195ccc866e1c11e1fd29e8ae7520399f4 Mon Sep 17 00:00:00 2001 From: yirutang Date: Fri, 15 Jan 2021 13:27:28 -0800 Subject: [PATCH 5/6] . --- .../cloud/bigquery/storage/v1alpha2/StreamWriterTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java index 6a265c3910..c60750d48d 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java @@ -854,8 +854,7 @@ public void testFlushAll() throws Exception { ApiFuture appendFuture1 = sendTestMessage(writer, new String[] {"A"}); ApiFuture appendFuture2 = sendTestMessage(writer, new String[] {"B"}); ApiFuture appendFuture3 = sendTestMessage(writer, new String[] {"C"}); - - assertFalse(appendFuture3.isDone()); + writer.flushAll(100000); assertTrue(appendFuture3.isDone()); From b154fc4629b39954bbc76cb466930d436e02177c Mon Sep 17 00:00:00 2001 From: yirutang Date: Fri, 15 Jan 2021 13:33:32 -0800 Subject: [PATCH 6/6] . --- .../cloud/bigquery/storage/v1alpha2/StreamWriterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java index c60750d48d..7a96b87829 100644 --- a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java +++ b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/StreamWriterTest.java @@ -854,7 +854,7 @@ public void testFlushAll() throws Exception { ApiFuture appendFuture1 = sendTestMessage(writer, new String[] {"A"}); ApiFuture appendFuture2 = sendTestMessage(writer, new String[] {"B"}); ApiFuture appendFuture3 = sendTestMessage(writer, new String[] {"C"}); - + writer.flushAll(100000); assertTrue(appendFuture3.isDone());