@Override
  public CommitWorkStream commitWorkStream() {
    return windmillStreamFactory.createCommitWorkStream(
        dispatcherClient.getWindmillServiceStub(), throttleTimers.commitWorkThrottleTimer());
  }
@Test
  public void testStreamingCommit() throws Exception {
    ConcurrentHashMap<Long, WorkItemCommitRequest> commitRequests = new ConcurrentHashMap<>();
    serviceRegistry.addService(
        new CloudWindmillServiceV1Alpha1ImplBase() {
          @Override
          public StreamObserver<StreamingCommitWorkRequest> commitWorkStream(
              StreamObserver<StreamingCommitResponse> responseObserver) {
            return getTestCommitStreamObserver(responseObserver, commitRequests);
          }
        });

    // Make the commit requests, waiting for each of them to be verified and acknowledged.
    CommitWorkStream stream = client.commitWorkStream();
    commitWorkTestHelper(stream, commitRequests, 0, 500);
    stream.halfClose();
    assertTrue(stream.awaitTermination(30, TimeUnit.SECONDS));
  }