@Override
  public CommitWorkStream commitWorkStream() {
    return windmillStreamFactory.createCommitWorkStream(
        dispatcherClient.getWindmillServiceStub(), throttleTimers.commitWorkThrottleTimer());
  }
@Test
  public void testStreamingCommitManyThreads() throws Exception {
    ConcurrentHashMap<Long, WorkItemCommitRequest> commitRequests = new ConcurrentHashMap<>();
    serviceRegistry.addService(
        new CloudWindmillServiceV1Alpha1ImplBase() {
          @Override
          public StreamObserver<StreamingCommitWorkRequest> commitWorkStream(
              StreamObserver<StreamingCommitResponse> responseObserver) {
            return getTestCommitStreamObserver(responseObserver, commitRequests);
          }
        });
    ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);
    // Make the commit requests, waiting for each of them to be verified and acknowledged.
    CommitWorkStream stream = client.commitWorkStream();
    List<Future<?>> futures = new ArrayList<>();
    for (int i = 0; i < 10; ++i) {
      final int startRequestId = i * 50;
      futures.add(
          executor.submit(
              () -> {
                try {
                  commitWorkTestHelper(stream, commitRequests, startRequestId, 50);
                } catch (InterruptedException e) {
                  throw new RuntimeException(e);
                }
              }));
    }
    // Surface any exceptions that might be thrown by submitting by blocking on the future.
    for (Future<?> f : futures) {
      f.get();
    }
    stream.halfClose();
    assertTrue(stream.awaitTermination(30, TimeUnit.SECONDS));
    executor.shutdown();
  }