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 verifyCacheTokensAreUsedInNewBundleRequest() throws InterruptedException {
    BeamFnDataOutboundAggregator mockInputSender = mock(BeamFnDataOutboundAggregator.class);
    when(fnApiControlClient.handle(any(BeamFnApi.InstructionRequest.class)))
        .thenReturn(
            CompletableFuture.<InstructionResponse>completedFuture(
                InstructionResponse.newBuilder().build()));

    ProcessBundleDescriptor descriptor1 =
        ProcessBundleDescriptor.newBuilder().setId("descriptor1").build();

    List<RemoteInputDestination> remoteInputs =
        Collections.singletonList(
            RemoteInputDestination.of(
                FullWindowedValueCoder.of(VarIntCoder.of(), GlobalWindow.Coder.INSTANCE),
                SDK_GRPC_READ_TRANSFORM));

    BundleProcessor processor1 = sdkHarnessClient.getProcessor(descriptor1, remoteInputs);
    when(dataService.createOutboundAggregator(any(), anyBoolean())).thenReturn(mockInputSender);

    StateRequestHandler stateRequestHandler = Mockito.mock(StateRequestHandler.class);
    List<BeamFnApi.ProcessBundleRequest.CacheToken> cacheTokens =
        Collections.singletonList(
            BeamFnApi.ProcessBundleRequest.CacheToken.newBuilder().getDefaultInstanceForType());
    when(stateRequestHandler.getCacheTokens()).thenReturn(cacheTokens);

    processor1.newBundle(
        ImmutableMap.of(
            SDK_GRPC_WRITE_TRANSFORM,
            RemoteOutputReceiver.of(ByteArrayCoder.of(), mock(FnDataReceiver.class))),
        stateRequestHandler,
        BundleProgressHandler.ignored());

    // Retrieve the requests made to the FnApiControlClient
    ArgumentCaptor<BeamFnApi.InstructionRequest> reqCaptor =
        ArgumentCaptor.forClass(BeamFnApi.InstructionRequest.class);
    Mockito.verify(fnApiControlClient, Mockito.times(1)).handle(reqCaptor.capture());
    List<BeamFnApi.InstructionRequest> requests = reqCaptor.getAllValues();

    // Verify that the cache tokens are included in the ProcessBundleRequest
    assertThat(
        requests.get(0).getRequestCase(),
        is(BeamFnApi.InstructionRequest.RequestCase.PROCESS_BUNDLE));
    assertThat(requests.get(0).getProcessBundle().getCacheTokensList(), is(cacheTokens));
  }