@Benchmark
  @Threads(16) // Use several threads since we expect contention during bundle processing.
  public void testLargeBundle(TrivialTransform trivialTransform) throws Exception {
    Map<String, ? super Coder<WindowedValue<?>>> remoteOutputCoders =
        trivialTransform.descriptor.getRemoteOutputCoders();
    Map<String, RemoteOutputReceiver<?>> outputReceivers = new HashMap<>();
    AtomicInteger outputValuesCount = new AtomicInteger();
    for (Entry<String, ? super Coder<WindowedValue<?>>> remoteOutputCoder :
        remoteOutputCoders.entrySet()) {
      outputReceivers.put(
          remoteOutputCoder.getKey(),
          RemoteOutputReceiver.of(
              (Coder) remoteOutputCoder.getValue(),
              (FnDataReceiver<? super WindowedValue<?>>)
                  (WindowedValue<?> value) -> outputValuesCount.incrementAndGet()));
    }
    try (RemoteBundle bundle =
        trivialTransform.processor.newBundle(outputReceivers, BundleProgressHandler.ignored())) {
      for (int i = 0; i < 1_000; i++) {
        Iterables.getOnlyElement(bundle.getInputReceivers().values())
            .accept(valueInGlobalWindow(new byte[0]));
      }
    }
    assertEquals(3_000, outputValuesCount.getAndSet(0));
  }
@Test
  public void testLargeBundle() throws Exception {
    TrivialTransform transform = new TrivialTransform();
    transform.elementsEmbedding = elementsEmbedding;
    new ProcessBundleBenchmark().testLargeBundle(transform);
    transform.tearDown();
  }