@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();
    }
  }
@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));
  }