fix: add client shutdown if request waiting in request queue for too long. #2017

GaoleMeng · 2023-02-24T09:12:56Z

this is needed as offline investigation, it's possible for grpc streaming lib to never trigger doneCallback / requestCallback under some scenario, e.g. when there's too many GAX threads. We should add this timeout to help connection spit out the requests on dead connections

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
Ensure the tests and linter pass
Code coverage does not decrease (if any source code was changed)
Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the samples format.

prerequisite for multiplexing client

new stream name as a switch of destinationt

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

also fixed a tiny bug inside fake bigquery write impl for getting thre response from offset

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

…nt or not

possible the proto schema does not contain this field

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

reuvenlax · 2023-02-24T23:21:36Z

...oud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

+    Duration milliSinceLastCallback =
+        Duration.between(lastRequestCallbackTriggerTime, Instant.now());
+    if (milliSinceLastCallback.compareTo(MAXIMUM_REQUEST_CALLBACK_WAIT_TIME) > 0) {
+      throw new IllegalStateException(


I don't think this is really an IllegalStateException

Let's use a generic runtime exception, the goal is only to trigger uncaught exception handler

reuvenlax · 2023-02-24T23:22:39Z

...oud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

@@ -659,6 +678,18 @@ private void appendLoop() {
    log.info("Append thread is done. Stream: " + streamName + " id: " + writerId);
  }

+  private void throwIfWaitCallbackTooLong() {
+    Duration milliSinceLastCallback =
+        Duration.between(lastRequestCallbackTriggerTime, Instant.now());


If a StreamWriter is created and then there is a pause before using it, won't this trigger? (odd scenario, I know). Actually if a user ever pauses for 10 minutes, won't this cause the exception the be thrown the next time they use the StreamWriter?

The caller of the throwIfWaitCallbackTooLong is checking on the inflight queue size before triggering this check

Alternatively, would it make sense to store a timestamp in the RequestWrapper itself when it is added to the inflightRequestQueue? Since the queue is in order, you then can just check the timestamp at the head of the queue

right, but append will add to this queue, right? So if append is called after a long delay, the appendLoop might hit this condition (if the appendLoop runs before the callback comes back).

Changed to a storing timestamp directly during creation of request wrapper

reuvenlax · 2023-02-25T07:39:56Z

...oud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

+  private void throwIfWaitCallbackTooLong(Instant timeToCheck) {
+    Duration milliSinceLastCallback = Duration.between(timeToCheck, Instant.now());
+    if (milliSinceLastCallback.compareTo(MAXIMUM_REQUEST_CALLBACK_WAIT_TIME) > 0) {
+      throw new RuntimeException(


Are you sure that a RuntimeException will be caught and will cancel things?

yeah: https://github.com/googleapis/java-bigquerystorage/blob/main/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java#L249

reuvenlax · 2023-02-25T07:43:01Z

...oud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

    AppendRequestAndResponse(AppendRowsRequest message, StreamWriter streamWriter) {
      this.appendResult = SettableApiFuture.create();
      this.message = message;
      this.messageSize = message.getProtoRows().getSerializedSize();
      this.streamWriter = streamWriter;
+      this.requestCreationTimeStamp = Instant.now();


Not sure if this is 100% right, since this gets called when the request is added to waitingRequestQueue. I think we want a timestamp that is set close to when the actual RPC goes out.

I was trying to start counting time when user call Append. But let's change to set the timestamp when the request is added to the inflight queue.

The idea is to detect stuck RPCs. If things are stuck in the waitingRequestQueue (and nothing is stuck in the inflight queue), then presumably something else is very wrong.

yirutang · 2023-02-27T20:32:24Z

...oud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

+   * We will constantly checking how much time we have been waiting for the next request callback
+   * if we wait too much time we will start shutting down the connections and clean up the queues.
+   */
+  private static Duration MAXIMUM_REQUEST_CALLBACK_WAIT_TIME = Duration.ofMinutes(10);


To keep it safe, let's put a longer time say 30 minutes.

Let's use 20 minutes, 30 minutes sound crazy long for a pipeline to stop

Right now the user waits forever on the queue, I am just afraid some user does crazy things. I am fine with 20 min.

GaoleMeng and others added 30 commits September 13, 2022 01:58

feat: Split writer into connection worker and wrapper, this is a

5a63d95

prerequisite for multiplexing client

feat: add connection worker pool skeleton, used for multiplexing client

5a13302

Merge branch 'main' into main

0297204

feat: add Load api for connection worker for multiplexing client

8a81ad3

Merge remote-tracking branch 'upstream/main'

68fd040

Merge remote-tracking branch 'upstream/main'

3106dae

Merge branch 'googleapis:main' into main

5bf04e5

Merge branch 'main' of https://github.com/GaoleMeng/java-bigquerystorage

2fc7551

feat: add multiplexing support to connection worker. We will treat every

7a6d919

new stream name as a switch of destinationt

🦉 Updates from OwlBot post-processor

3ba7659

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Updates from OwlBot post-processor

f379a78

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Merge branch 'main' of https://github.com/GaoleMeng/java-bigquerystorage

9307776

🦉 Updates from OwlBot post-processor

de73013

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

feat: port the multiplexing client core algorithm and basic tests

19005a1

also fixed a tiny bug inside fake bigquery write impl for getting thre response from offset

Merge branch 'googleapis:main' into main

c5d14ba

🦉 Updates from OwlBot post-processor

644360a

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Merge branch 'googleapis:main' into main

3099d82

Merge branch 'googleapis:main' into main

e707dd6

Merge branch 'main' of https://github.com/GaoleMeng/java-bigquerystorage

9e7a8fa

Merge branch 'googleapis:main' into main

31f1755

feat: wire multiplexing connection pool to stream writer

44c36fc

feat: some fixes for multiplexing client

87a4036

Merge remote-tracking branch 'upstream/main'

c92ea1b

Merge branch 'googleapis:main' into main

019520c

feat: fix some todos, and reject the mixed behavior of passed in clie…

47893df

…nt or not

Merge remote-tracking branch 'upstream/main'

8bd4e6a

Merge remote-tracking branch 'upstream/main'

83409b0

Merge branch 'googleapis:main' into main

f7dd72d

Merge branch 'googleapis:main' into main

a48399f

feat: fix the bug that we may peek into the write_stream field but it's

6789bc9

possible the proto schema does not contain this field

GaoleMeng and others added 4 commits February 8, 2023 13:33

Merge branch 'googleapis:main' into main

5a74e5b

Merge branch 'main' of https://github.com/GaoleMeng/java-bigquerystorage

bafc1b4

🦉 Updates from OwlBot post-processor

60f7ad3

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Merge branch 'googleapis:main' into main

d1dbb59

GaoleMeng requested review from a team and shollyman February 24, 2023 09:12

product-auto-label bot added size: m Pull request size is medium. api: bigquerystorage Issues related to the googleapis/java-bigquerystorage API. labels Feb 24, 2023

GaoleMeng requested a review from yirutang February 24, 2023 09:13

reuvenlax reviewed Feb 24, 2023

View reviewed changes

reuvenlax reviewed Feb 25, 2023

View reviewed changes

yirutang reviewed Feb 27, 2023

View reviewed changes

yirutang approved these changes Feb 28, 2023

View reviewed changes

GaoleMeng added the owlbot:run Add this label to trigger the Owlbot post processor. label Feb 28, 2023

gcf-owl-bot bot removed the owlbot:run Add this label to trigger the Owlbot post processor. label Feb 28, 2023

GaoleMeng added the owlbot:run Add this label to trigger the Owlbot post processor. label Feb 28, 2023

gcf-owl-bot bot removed the owlbot:run Add this label to trigger the Owlbot post processor. label Feb 28, 2023

GaoleMeng added the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Feb 28, 2023

yoshi-kokoro removed the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Feb 28, 2023

product-auto-label bot added size: s Pull request size is small. and removed size: m Pull request size is medium. labels Feb 28, 2023

fix: add client side timeout if inflight request wait too long

28a4d82

product-auto-label bot added size: m Pull request size is medium. and removed size: s Pull request size is small. labels Feb 28, 2023

Merge branch 'main' into main

fab5422

GaoleMeng added the owlbot:run Add this label to trigger the Owlbot post processor. label Mar 1, 2023

gcf-owl-bot bot removed the owlbot:run Add this label to trigger the Owlbot post processor. label Mar 1, 2023

GaoleMeng merged commit 91da88b into googleapis:main Mar 1, 2023

release-please bot mentioned this pull request Mar 1, 2023

chore(main): release 2.33.0 #2022

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: add client shutdown if request waiting in request queue for too long. #2017

fix: add client shutdown if request waiting in request queue for too long. #2017

GaoleMeng commented Feb 24, 2023

reuvenlax Feb 24, 2023

GaoleMeng Feb 24, 2023

reuvenlax Feb 24, 2023

GaoleMeng Feb 24, 2023

reuvenlax Feb 24, 2023

reuvenlax Feb 24, 2023

GaoleMeng Feb 25, 2023

reuvenlax Feb 25, 2023

GaoleMeng Feb 27, 2023

reuvenlax Feb 25, 2023

GaoleMeng Feb 27, 2023

reuvenlax Feb 27, 2023

yirutang Feb 27, 2023

GaoleMeng Feb 28, 2023

yirutang Feb 28, 2023

fix: add client shutdown if request waiting in request queue for too long. #2017

fix: add client shutdown if request waiting in request queue for too long. #2017

Conversation

GaoleMeng commented Feb 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment