private void processElementWithSdkHarness(WindowedValue<InputT> element) throws Exception {
    checkState(
        stageBundleFactory != null, "%s not yet prepared", StageBundleFactory.class.getName());
    checkState(
        stateRequestHandler != null, "%s not yet prepared", StateRequestHandler.class.getName());
    OutputReceiverFactory receiverFactory =
        new OutputReceiverFactory() {
          @Override
          public FnDataReceiver<OutputT> create(String pCollectionId) {
            return (receivedElement) -> {
              // handover to queue, do not block the grpc thread
              outputQueue.put(KV.of(pCollectionId, receivedElement));
            };
          }
        };
    try (RemoteBundle bundle =
        stageBundleFactory.getBundle(receiverFactory, stateRequestHandler, progressHandler)) {
      logger.debug(String.format("Sending value: %s", element));
      // TODO(BEAM-4681): Add support to Flink to support portable timers.
      Iterables.getOnlyElement(bundle.getInputReceivers().values()).accept(element);
      // TODO: it would be nice to emit results as they arrive, can thread wait non-blocking?
    }
    // RemoteBundle close blocks until all results are received
    KV<String, OutputT> result;
    while ((result = outputQueue.poll()) != null) {
      outputManager.output(outputMap.get(result.getKey()), (WindowedValue) result.getValue());
    }
  }
private void processElementWithSdkHarness(WindowedValue<InputT> element) throws Exception {
    checkState(
        stageBundleFactory != null, "%s not yet prepared", StageBundleFactory.class.getName());
    checkState(
        stateRequestHandler != null, "%s not yet prepared", StateRequestHandler.class.getName());
    OutputReceiverFactory receiverFactory =
        new OutputReceiverFactory() {
          @Override
          public FnDataReceiver<OutputT> create(String pCollectionId) {
            return (receivedElement) -> {
             
              outputQueue.put(KV.of(pCollectionId, receivedElement));
            };
          }
        };
    try (RemoteBundle bundle =
        stageBundleFactory.getBundle(receiverFactory, stateRequestHandler, progressHandler)) {
      logger.debug(String.format("Sending value: %s", element));
     
      Iterables.getOnlyElement(bundle.getInputReceivers().values()).accept(element);
     
    }
   
    KV<String, OutputT> result;
    while ((result = outputQueue.poll()) != null) {
      outputManager.output(outputMap.get(result.getKey()), (WindowedValue) result.getValue());
    }
  }
private void processelementwithsdkharness(windowedvalue<inputt> element) throws exception { checkstate( stagebundlefactory != null, "%s not yet prepared", stagebundlefactory.class.getname()); checkstate( staterequesthandler != null, "%s not yet prepared", staterequesthandler.class.getname()); outputreceiverfactory receiverfactory = new outputreceiverfactory() { @override public fndatareceiver<outputt> create(string pcollectionid) { return (receivedelement) -> { outputqueue.put(kv.of(pcollectionid, receivedelement)); }; } }; try (remotebundle bundle = stagebundlefactory.getbundle(receiverfactory, staterequesthandler, progresshandler)) { logger.debug(string.format("sending value: %s", element)); iterables.getonlyelement(bundle.getinputreceivers().values()).accept(element); } kv<string, outputt> result; while ((result = outputqueue.poll()) != null) { outputmanager.output(outputmap.get(result.getkey()), (windowedvalue) result.getvalue()); } }