public BundleProcessor getProcessor(
      BeamFnApi.ProcessBundleDescriptor descriptor,
      List<RemoteInputDestination> remoteInputDesinations) {
    checkState(
        !descriptor.hasStateApiServiceDescriptor(),
        "The %s cannot support a %s containing a state %s.",
        BundleProcessor.class.getSimpleName(),
        BeamFnApi.ProcessBundleDescriptor.class.getSimpleName(),
        Endpoints.ApiServiceDescriptor.class.getSimpleName());
    return getProcessor(descriptor, remoteInputDesinations, NoOpStateDelegator.INSTANCE);
  }
@Test
  public void testBundleCheckpointCallback() throws Exception {
    BeamFnDataOutboundAggregator mockInputSender = mock(BeamFnDataOutboundAggregator.class);

    CompletableFuture<InstructionResponse> processBundleResponseFuture = new CompletableFuture<>();
    when(fnApiControlClient.handle(any(BeamFnApi.InstructionRequest.class)))
        .thenReturn(processBundleResponseFuture);

    FullWindowedValueCoder<String> coder =
        FullWindowedValueCoder.of(StringUtf8Coder.of(), Coder.INSTANCE);
    BundleProcessor processor =
        sdkHarnessClient.getProcessor(
            descriptor,
            Collections.singletonList(
                RemoteInputDestination.of(
                    (FullWindowedValueCoder) coder, SDK_GRPC_READ_TRANSFORM)));
    when(dataService.createOutboundAggregator(any(), anyBoolean())).thenReturn(mockInputSender);

    BundleProgressHandler mockProgressHandler = mock(BundleProgressHandler.class);
    BundleSplitHandler mockSplitHandler = mock(BundleSplitHandler.class);
    BundleCheckpointHandler mockCheckpointHandler = mock(BundleCheckpointHandler.class);
    BundleFinalizationHandler mockFinalizationHandler = mock(BundleFinalizationHandler.class);

    ProcessBundleResponse response =
        ProcessBundleResponse.newBuilder()
            .addResidualRoots(DelayedBundleApplication.getDefaultInstance())
            .build();
    try (ActiveBundle activeBundle =
        processor.newBundle(
            Collections.emptyMap(),
            Collections.emptyMap(),
            (request) -> {
              throw new UnsupportedOperationException();
            },
            mockProgressHandler,
            mockSplitHandler,
            mockCheckpointHandler,
            mockFinalizationHandler)) {
      processBundleResponseFuture.complete(
          InstructionResponse.newBuilder().setProcessBundle(response).build());
    }

    verify(mockProgressHandler).onCompleted(response);
    verify(mockCheckpointHandler).onCheckpoint(response);
    verifyNoMoreInteractions(mockFinalizationHandler, mockSplitHandler);
  }