FnApiControlClientPoolService extends BeamFnControlGrpc.BeamFnControlImplBase implements FnService { @Override public void getProcessBundleDescriptor( BeamFnApi.GetProcessBundleDescriptorRequest request, StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) { String bundleDescriptorId = request.getProcessBundleDescriptorId(); LOG.info("getProcessBundleDescriptor request with id {}", bundleDescriptorId); BeamFnApi.ProcessBundleDescriptor descriptor = processBundleDescriptors.get(bundleDescriptorId); if (descriptor == null) { String msg = String.format("ProcessBundleDescriptor with id %s not found", bundleDescriptorId); responseObserver.onError(new StatusException(Status.NOT_FOUND.withDescription(msg))); LOG.error(msg); } else { responseObserver.onNext(descriptor); responseObserver.onCompleted(); } } private  FnApiControlClientPoolService(
      ControlClientPool.Sink clientSink, HeaderAccessor headerAccessor); static FnApiControlClientPoolService offeringClientsToPool(
      ControlClientPool.Sink clientPool, HeaderAccessor headerAccessor); @Override StreamObserver<BeamFnApi.InstructionResponse> control(
      StreamObserver<BeamFnApi.InstructionRequest> requestObserver); @Override void getProcessBundleDescriptor(
      BeamFnApi.GetProcessBundleDescriptorRequest request,
      StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver); @Override void close();  }
@Test public void testUnknownBundle() throws Exception { BeamFnApi.GetProcessBundleDescriptorRequest request = BeamFnApi.GetProcessBundleDescriptorRequest.newBuilder() .setProcessBundleDescriptorId("missing") .build(); StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver = mock(StreamObserver.class); controlService.getProcessBundleDescriptor(request, responseObserver); verify(responseObserver) .onError( argThat( e -> e instanceof StatusException && ((StatusException) e).getStatus().getCode() == Status.Code.NOT_FOUND)); }