public void registerWithMaster() throws CancelledException, InternalException,
      DeadlineExceededException, InterruptedException {
    // onNext() - send the request batches
    // onError() - when an error occurs on the worker side, propagate the error status to
    //             the master side and then close on the worker side, the master will
    //             handle necessary cleanup on its side
    // onCompleted() - complete on the client side when all the batches are sent
    //                 and all ACKs are received
    mWorkerRequestObserver = mAsyncClient
        .withDeadlineAfter(mDeadlineMs, TimeUnit.MILLISECONDS)
        .registerWorkerStream(mMasterResponseObserver);
    try {
      registerInternal();
    } catch (DeadlineExceededException | InterruptedException | CancelledException e) {
      LOG.error("Worker {} - Error during the register stream, aborting now.", mWorkerId, e);
      // These exceptions are internal to the worker
      // Propagate to the master side so it can clean up properly
      mWorkerRequestObserver.onError(e);
      throw e;
    }
    // The only exception that is not propagated to the master is InternalException.
    // We assume that is from the master so there is no need to send it back again.
  }
@Test(timeout = 5000)
  public void registerSuccess() throws Exception {
    // store workerId to requestCount
    ConcurrentMap<Long, AtomicInteger> requestCount = new ConcurrentHashMap<>();
    // create a server that accepts and completes the registration properly
    createRegisterService(
        (responseObserver) -> new NoOpStreamObserver() {
              @Override
              public void onNext(RegisterWorkerPRequest value) {
                long workerId = value.getWorkerId();

                requestCount.putIfAbsent(workerId, new AtomicInteger(0));
                requestCount.get(workerId).incrementAndGet();

                responseObserver.onNext(RegisterWorkerPResponse.newBuilder().build());
              }

              @Override
              public void onError(Throwable t) {
              }

              @Override
              public void onCompleted() {
                responseObserver.onCompleted();
              }
            });

    // sends 5 requests for one workerId in total
    long workerId = 1L;
    RegisterStreamer streamer = createMockRegisterStreamer(mChannel, workerId, 5);
    streamer.registerWithMaster();
    assertEquals(1, requestCount.size());
    assertEquals(5, requestCount.get(workerId).get());
  }