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 testProcessingOfTheLastInput() throws CannotProvideCoderException {
    Event[] events = {
      Event.create(0, "id-1", "a"),
      Event.create(1, "id-1", "b"),
      Event.create(2, "id-1", StringEventExaminer.LAST_INPUT)
    };

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

    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", "ab" + StringEventExaminer.LAST_INPUT));

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