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 testPerfectOrderingProcessing() throws CannotProvideCoderException {
    Event[] events = {
      Event.create(0, "id-1", "a"),
      Event.create(1, "id-1", "b"),
      Event.create(2, "id-1", "c"),
      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(
                3L,
                0,
                null,
                null,
                4,
                Arrays.stream(events).filter(e -> e.getKey().equals("id-1")).count(),
                0,
                false)));
    expectedStatuses.add(
        KV.of(
            "id-2",
            OrderedProcessingStatus.create(
                1L,
                0,
                null,
                null,
                2,
                Arrays.stream(events).filter(e -> e.getKey().equals("id-2")).count(),
                0,
                false)));

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

    testProcessing(
        events,
        expectedStatuses,
        expectedOutput,
        EMISSION_FREQUENCY_ON_EVERY_ELEMENT,
        INITIAL_SEQUENCE_OF_0,
        LARGE_MAX_RESULTS_PER_OUTPUT,
        DONT_PRODUCE_STATUS_ON_EVERY_EVENT);
  }