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 testNewBundleAndProcessElements() throws Exception {
    SdkHarnessClient client = harness.client();
    BundleProcessor processor =
        client.getProcessor(
            descriptor,
            Collections.singletonList(
                RemoteInputDestination.of(
                    (FullWindowedValueCoder)
                        FullWindowedValueCoder.of(StringUtf8Coder.of(), Coder.INSTANCE),
                    SDK_GRPC_READ_TRANSFORM)));

    Collection<WindowedValue<String>> outputs = new ArrayList<>();
    try (RemoteBundle activeBundle =
        processor.newBundle(
            Collections.singletonMap(
                SDK_GRPC_WRITE_TRANSFORM,
                RemoteOutputReceiver.of(
                    FullWindowedValueCoder.of(
                        LengthPrefixCoder.of(StringUtf8Coder.of()), Coder.INSTANCE),
                    outputs::add)),
            BundleProgressHandler.ignored())) {
      FnDataReceiver<WindowedValue<?>> bundleInputReceiver =
          Iterables.getOnlyElement(activeBundle.getInputReceivers().values());
      bundleInputReceiver.accept(WindowedValue.valueInGlobalWindow("foo"));
      bundleInputReceiver.accept(WindowedValue.valueInGlobalWindow("bar"));
      bundleInputReceiver.accept(WindowedValue.valueInGlobalWindow("baz"));
    }

    // The bundle can be a simple function of some sort, but needs to be complete.
    assertThat(
        outputs,
        containsInAnyOrder(
            WindowedValue.valueInGlobalWindow("spam"),
            WindowedValue.valueInGlobalWindow("ham"),
            WindowedValue.valueInGlobalWindow("eggs")));
  }