feat: Change one thread per retry to use a thread pool (#1898)

* feat: Split writer into connection worker and wrapper, this is a prerequisite for multiplexing client * feat: add connection worker pool skeleton, used for multiplexing client * feat: add Load api for connection worker for multiplexing client * feat: add multiplexing support to connection worker. We will treat every new stream name as a switch of destinationt * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * 🦉 Updates from OwlBot post-processor 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 also fixed a tiny bug inside fake bigquery write impl for getting thre response from offset * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: wire multiplexing connection pool to stream writer * feat: some fixes for multiplexing client * feat: fix some todos, and reject the mixed behavior of passed in client or not * feat: fix the bug that we may peek into the write_stream field but it's possible the proto schema does not contain this field * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: fix the bug that we may peek into the write_stream field but it's possible the proto schema does not contain this field * feat: add getInflightWaitSeconds implementation * feat: Add schema comparision in connection loop to ensure schema update for the same stream name can be notified * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: add schema update support to multiplexing * fix: fix windows build bug: windows Instant resolution is different with linux * fix: fix another failing tests for windows build * fix: fix another test failure for Windows build * feat: Change new thread for each retry to be a thread pool to avoid create/tear down too much threads if lots of retries happens * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

Author: GaoleMeng

samples/snippets/src/main/java/com/example/bigquerystorage/WriteToDefaultStream.java

@@ -41,6 +41,8 @@
 import io.grpc.Status.Code;
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Phaser;
 import javax.annotation.concurrent.GuardedBy;
 import org.json.JSONArray;
@@ -193,6 +195,8 @@ static class AppendCompleteCallback implements ApiFutureCallback<AppendRowsRespo
 
  private final DataWriter parent;
  private final AppendContext appendContext;
+ // Prepare a thread pool
+ static ExecutorService pool = Executors.newFixedThreadPool(50);
 
  public AppendCompleteCallback(DataWriter parent, AppendContext appendContext) {
  this.parent = parent;
@@ -213,19 +217,18 @@ public void onFailure(Throwable throwable) {
  && RETRIABLE_ERROR_CODES.contains(status.getCode())) {
  appendContext.retryCount++;
  // Use a separate thread to avoid potentially blocking while we are in a callback.
- new Thread(
- () -> {
- try {
- // Since default stream appends are not ordered, we can simply retry the
- // appends.
- // Retrying with exclusive streams requires more careful consideration.
- this.parent.append(appendContext);
- } catch (Exception e) {
- // Fall through to return error.
- System.out.format("Failed to retry append: %s%n", e);
- }
- })
- .start();
+ pool.submit(
+ () -> {
+ try {
+ // Since default stream appends are not ordered, we can simply retry the
+ // appends.
+ // Retrying with exclusive streams requires more careful consideration.
+ this.parent.append(appendContext);
+ } catch (Exception e) {
+ // Fall through to return error.
+ System.out.format("Failed to retry append: %s%n", e);
+ }
+ });
  // Mark the existing attempt as done since it's being retried.
  done();
  return;
@@ -251,15 +254,14 @@ public void onFailure(Throwable throwable) {
  // Retry the remaining valid rows, but using a separate thread to
  // avoid potentially blocking while we are in a callback.
  if (dataNew.length() > 0) {
- new Thread(
- () -> {
- try {
- this.parent.append(new AppendContext(dataNew, 0));
- } catch (Exception e2) {
- System.out.format("Failed to retry append with filtered rows: %s%n", e2);
- }
- })
- .start();
+ pool.submit(
+ () -> {
+ try {
+ this.parent.append(new AppendContext(dataNew, 0));
+ } catch (Exception e2) {
+ System.out.format("Failed to retry append with filtered rows: %s%n", e2);
+ }
+ });
  }
  return;
  }