private Object getKey(WindowedValue<InT> elem) {
    KV<?, ?> kv = (KV<?, ?>) elem.getValue();
    if (kv == null) {
      return NULL_KEY;
    } else {
      Object key = kv.getKey();
      return key == null ? NULL_KEY : key;
    }
  }
@Test
  @Ignore("https://github.com/apache/beam/issues/23745")
  public void testPipelineWithAggregation() {
    final List<KV<String, Long>> input =
        new ArrayList<>(
            Arrays.asList(
                KV.of("apple", 2L),
                KV.of("banana", 5L),
                KV.of("apple", 8L),
                KV.of("grape", 10L),
                KV.of("banana", 5L)));

    // TODO: remove after SAMZA-2761 fix
    for (int i = 0; i < 50; i++) {
      input.add(KV.of("*", 0L));
    }

    PCollection<KV<String, Long>> sums =
        pipeline
            .apply(Create.of(input))
            .apply(Filter.by(x -> !x.getKey().equals("*")))
            .apply(Sum.longsPerKey());

    PAssert.that(sums)
        .containsInAnyOrder(
            Arrays.asList(KV.of("apple", 10L), KV.of("banana", 10L), KV.of("grape", 10L)));

    pipeline.run();
  }