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 testPipelineWithState() {
    final List<KV<String, String>> input =
        new ArrayList<>(
            Arrays.asList(
                KV.of("apple", "red"),
                KV.of("banana", "yellow"),
                KV.of("apple", "yellow"),
                KV.of("grape", "purple"),
                KV.of("banana", "yellow")));
    final Map<String, Integer> expectedCount = ImmutableMap.of("apple", 2, "banana", 2, "grape", 1);

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

    final DoFn<KV<String, String>, KV<String, Integer>> fn =
        new DoFn<KV<String, String>, KV<String, Integer>>() {

          @StateId("cc")
          private final StateSpec<CombiningState<Integer, int[], Integer>> countState =
              StateSpecs.combiningFromInputInternal(VarIntCoder.of(), Sum.ofIntegers());

          @ProcessElement
          public void processElement(
              ProcessContext c, @StateId("cc") CombiningState<Integer, int[], Integer> countState) {

            if (c.element().getKey().equals("*")) {
              return;
            }

            countState.add(1);
            String key = c.element().getKey();
            int n = countState.read();
            if (n >= expectedCount.get(key)) {
              c.output(KV.of(key, n));
            }
          }
        };

    PCollection<KV<String, Integer>> counts = pipeline.apply(Create.of(input)).apply(ParDo.of(fn));

    PAssert.that(counts)
        .containsInAnyOrder(
            expectedCount.entrySet().stream()
                .map(entry -> KV.of(entry.getKey(), entry.getValue()))
                .collect(Collectors.toList()));

    pipeline.run();
  }