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 testRegisterWithStateRequiresStateDelegator() throws Exception {
    ProcessBundleDescriptor descriptor =
        ProcessBundleDescriptor.newBuilder()
            .setId("test")
            .setStateApiServiceDescriptor(ApiServiceDescriptor.newBuilder().setUrl("foo"))
            .build();

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

    thrown.expect(IllegalStateException.class);
    thrown.expectMessage("containing a state");
    sdkHarnessClient.getProcessor(descriptor, remoteInputs);
  }