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 testSequenceGapProcessingInBufferedOutput() throws CannotProvideCoderException {
    int maxResultsPerOutput = 3;

    long[] sequences = new long[] {2, 3, 7, 8, 9, 10, 1, 4, 5, 6};

    List<Event> events = new ArrayList<>(sequences.length);
    List<KV<String, String>> expectedOutput = new ArrayList<>(sequences.length);

    StringBuilder output = new StringBuilder();
    String outputPerElement = ".";
    String key = "id-1";

    for (long sequence : sequences) {
      events.add(Event.create(sequence, key, outputPerElement));
      output.append(outputPerElement);
      expectedOutput.add(KV.of(key, output.toString()));
    }

    int numberOfReceivedEvents = 0;
    Collection<KV<String, OrderedProcessingStatus>> expectedStatuses = new ArrayList<>();

    // First elements are out-of-sequence and they just get buffered. Earliest and latest sequence
    // numbers keep changing.
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 1, 2L, 2L, ++numberOfReceivedEvents, 0L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 2, 2L, 3L, ++numberOfReceivedEvents, 0L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 3, 2L, 7L, ++numberOfReceivedEvents, 0L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 4, 2L, 8L, ++numberOfReceivedEvents, 0L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 5, 2L, 9L, ++numberOfReceivedEvents, 0L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                null, 6, 2L, 10L, ++numberOfReceivedEvents, 0L, 0, false)));
    // --- 1 has appeared and caused the batch to be sent out.
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                3L, 4, 7L, 10L, ++numberOfReceivedEvents, 3L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                4L, 4, 7L, 10L, ++numberOfReceivedEvents, 4L, 0, false)));
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                5L, 4, 7L, 10L, ++numberOfReceivedEvents, 5L, 0, false)));
    // --- 6 came and 6, 7, and 8 got output
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                8L, 2, 9L, 10L, ++numberOfReceivedEvents, 8L, 0, false)));
    // Last timer run produces the final status. Number of received events doesn't
    // increase,
    // this is the result of a timer processing
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                10L, 0, null, null, numberOfReceivedEvents, 10L, 0, false)));

    testProcessing(
        events.toArray(new Event[events.size()]),
        expectedStatuses,
        expectedOutput,
        EMISSION_FREQUENCY_ON_EVERY_ELEMENT,
        1L /* This dataset assumes 1 as the starting sequence */,
        maxResultsPerOutput,
        PRODUCE_STATUS_ON_EVERY_EVENT);
  }