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 handleCleanupWithStateWhenInputSenderFails() throws Exception {
    RuntimeException testException = new RuntimeException();

    BeamFnDataOutboundAggregator mockInputSender = mock(BeamFnDataOutboundAggregator.class);

    StateDelegator mockStateDelegator = mock(StateDelegator.class);
    StateDelegator.Registration mockStateRegistration = mock(StateDelegator.Registration.class);
    when(mockStateDelegator.registerForProcessBundleInstructionId(any(), any()))
        .thenReturn(mockStateRegistration);
    StateRequestHandler mockStateHandler = mock(StateRequestHandler.class);
    when(mockStateHandler.getCacheTokens()).thenReturn(Collections.emptyList());
    BundleProgressHandler mockProgressHandler = mock(BundleProgressHandler.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)),
            mockStateDelegator);
    when(dataService.createOutboundAggregator(any(), anyBoolean())).thenReturn(mockInputSender);

    doThrow(testException)
        .when(mockInputSender)
        .sendOrCollectBufferedDataAndFinishOutboundStreams();

    try {
      try (RemoteBundle activeBundle =
          processor.newBundle(
              ImmutableMap.of(
                  SDK_GRPC_WRITE_TRANSFORM,
                  RemoteOutputReceiver.of(ByteArrayCoder.of(), mock(FnDataReceiver.class))),
              mockStateHandler,
              mockProgressHandler)) {
        // We shouldn't be required to complete the process bundle response future.
      }
      fail("Exception expected");
    } catch (Exception e) {
      assertEquals(testException, e);

      verify(mockStateRegistration).abort();
      // We expect that we don't register the receiver and the next accept call will raise an error
      // making the data service aware of the error.
      verify(dataService, never()).unregisterReceiver(any());
      assertThrows(
          "Inbound observer closed.",
          Exception.class,
          () -> outputReceiverCaptor.getValue().accept(Elements.getDefaultInstance()));
    }
  }