public static <
          EventTypeT,
          EventKeyTypeT,
          ResultTypeT,
          StateTypeT extends MutableState<EventTypeT, ResultTypeT>>
      OrderedEventProcessor<EventTypeT, EventKeyTypeT, ResultTypeT, StateTypeT> create(
          OrderedProcessingHandler<EventTypeT, EventKeyTypeT, StateTypeT, ResultTypeT> handler) {
    return new AutoValue_OrderedEventProcessor<>(handler);
  }
@Test
  public void testUnfinishedProcessing() throws CannotProvideCoderException {
    Event[] events = {
      Event.create(2, "id-1", "c"),
      //   Excluded                     Event.create(1, "id-1", "b"),
      Event.create(0, "id-1", "a"),
      Event.create(3, "id-1", "d"),
      Event.create(0, "id-2", "a"),
      Event.create(1, "id-2", "b"),
    };

    Collection<KV<String, OrderedProcessingStatus>> expectedStatuses = new ArrayList<>();
    expectedStatuses.add(
        KV.of("id-1", OrderedProcessingStatus.create(0L, 2, 2L, 3L, 3, 1L, 0, false)));
    expectedStatuses.add(
        KV.of("id-2", OrderedProcessingStatus.create(1L, 0, null, null, 2, 2L, 0, false)));

    Collection<KV<String, String>> expectedOutput = new ArrayList<>();
    expectedOutput.add(KV.of("id-1", "a"));
    expectedOutput.add(KV.of("id-2", "a"));
    expectedOutput.add(KV.of("id-2", "ab"));

    testProcessing(events, expectedStatuses, expectedOutput, 1, 0, 1000, false);
  }