@Override
  public BeamFnDataOutboundAggregator createOutboundAggregator(
      ApiServiceDescriptor apiServiceDescriptor,
      Supplier<String> processBundleRequestIdSupplier,
      boolean collectElementsIfNoFlushes) {
    return new BeamFnDataOutboundAggregator(
        options,
        processBundleRequestIdSupplier,
        getClientFor(apiServiceDescriptor).getOutboundObserver(),
        collectElementsIfNoFlushes);
  }
@Test
  public void testForOutboundConsumer() throws Exception {
    CountDownLatch waitForInboundServerValuesCompletion = new CountDownLatch(2);
    Collection<BeamFnApi.Elements> inboundServerValues = new ConcurrentLinkedQueue<>();
    CallStreamObserver<BeamFnApi.Elements> inboundServerObserver =
        TestStreams.withOnNext(
                (BeamFnApi.Elements t) -> {
                  inboundServerValues.add(t);
                  waitForInboundServerValuesCompletion.countDown();
                })
            .build();

    Endpoints.ApiServiceDescriptor apiServiceDescriptor =
        Endpoints.ApiServiceDescriptor.newBuilder()
            .setUrl(this.getClass().getName() + "-" + UUID.randomUUID())
            .build();
    Server server =
        InProcessServerBuilder.forName(apiServiceDescriptor.getUrl())
            .addService(
                new BeamFnDataGrpc.BeamFnDataImplBase() {
                  @Override
                  public StreamObserver<BeamFnApi.Elements> data(
                      StreamObserver<BeamFnApi.Elements> outboundObserver) {
                    return inboundServerObserver;
                  }
                })
            .build();
    server.start();
    try {
      ManagedChannel channel =
          InProcessChannelBuilder.forName(apiServiceDescriptor.getUrl()).build();

      BeamFnDataGrpcClient clientFactory =
          new BeamFnDataGrpcClient(
              PipelineOptionsFactory.fromArgs(
                      new String[] {"--experiments=data_buffer_size_limit=20"})
                  .create(),
              (Endpoints.ApiServiceDescriptor descriptor) -> channel,
              OutboundObserverFactory.trivial());
      BeamFnDataOutboundAggregator aggregator =
          clientFactory.createOutboundAggregator(
              apiServiceDescriptor, () -> INSTRUCTION_ID_A, false);
      FnDataReceiver<WindowedValue<String>> fnDataReceiver =
          aggregator.registerOutputDataLocation(TRANSFORM_ID_A, CODER);
      fnDataReceiver.accept(valueInGlobalWindow("ABC"));
      fnDataReceiver.accept(valueInGlobalWindow("DEF"));
      fnDataReceiver.accept(valueInGlobalWindow("GHI"));
      aggregator.sendOrCollectBufferedDataAndFinishOutboundStreams();
      waitForInboundServerValuesCompletion.await();

      assertThat(inboundServerValues, contains(ELEMENTS_A_1, ELEMENTS_A_2));
    } finally {
      server.shutdownNow();
    }
  }