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

    // Array of sequences starting with 2 and the last element - 1.
    // Output will be buffered until the last event arrives
    long[] sequences = new long[maxResultsPerOutput * 3];
    for (int i = 0; i < sequences.length - 1; i++) {
      sequences[i] = i + 2L;
    }
    sequences[sequences.length - 1] = 1;

    List<Event> events = new ArrayList<>(sequences.length);
    Collection<KV<String, String>> expectedOutput = new ArrayList<>(sequences.length);
    Collection<KV<String, OrderedProcessingStatus>> expectedStatuses =
        new ArrayList<>(sequences.length + 10);

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

    int bufferedEventCount = 0;

    for (long sequence : sequences) {
      ++bufferedEventCount;

      events.add(Event.create(sequence, key, outputPerElement));
      output.append(outputPerElement);
      expectedOutput.add(KV.of(key, output.toString()));

      if (bufferedEventCount < sequences.length) {
        // Last event will result in a batch of events being produced. That's why it's excluded
        // here.
        expectedStatuses.add(
            KV.of(
                key,
                OrderedProcessingStatus.create(
                    null, bufferedEventCount, 2L, sequence, bufferedEventCount, 0L, 0, false)));
      }
    }

    // Statuses produced by the batched processing
    for (int i = maxResultsPerOutput; i < sequences.length; i += maxResultsPerOutput) {
      long lastOutputSequence = i;
      expectedStatuses.add(
          KV.of(
              key,
              OrderedProcessingStatus.create(
                  lastOutputSequence,
                  sequences.length - lastOutputSequence,
                  lastOutputSequence + 1,
                  (long) sequences.length,
                  sequences.length,
                  lastOutputSequence,
                  0,
                  false)));
    }

    // -- Final status - indicates that everything has been fully processed
    expectedStatuses.add(
        KV.of(
            key,
            OrderedProcessingStatus.create(
                (long) sequences.length,
                0,
                null,
                null,
                sequences.length,
                sequences.length,
                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);
  }