@Override
  public <K, T> UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output) {
    return UncommittedImmutableListBundle.create(output, key);
  }
@Test
  public void createKeyedBundleKeyed() {
    CommittedBundle<KV<String, Integer>> keyedBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("foo", StringUtf8Coder.of()), downstream)
            .commit(Instant.now());
    assertThat(keyedBundle.getKey().getKey(), equalTo("foo"));
  }