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 testWindowedProcessing() throws CannotProvideCoderException {

    Instant base = new Instant(0);
    TestStream<Event> values =
        TestStream.create(streamingPipeline.getCoderRegistry().getCoder(Event.class))
            .advanceWatermarkTo(base)
            .addElements(
                // Start of first window
                TimestampedValue.of(
                    Event.create(0, "id-1", "a"), base.plus(Duration.standardSeconds(1))),
                TimestampedValue.of(
                    Event.create(1, "id-1", "b"), base.plus(Duration.standardSeconds(2))),
                TimestampedValue.of(
                    Event.create(0, "id-2", "x"), base.plus(Duration.standardSeconds(1))),
                TimestampedValue.of(
                    Event.create(1, "id-2", "y"), base.plus(Duration.standardSeconds(2))),
                TimestampedValue.of(
                    Event.create(2, "id-2", "z"), base.plus(Duration.standardSeconds(2))),

                // Start of second window. Numbering must start with 0 again.
                TimestampedValue.of(
                    Event.create(0, "id-1", "c"), base.plus(Duration.standardSeconds(10))),
                TimestampedValue.of(
                    Event.create(1, "id-1", "d"), base.plus(Duration.standardSeconds(11))))
            .advanceWatermarkToInfinity();

    Pipeline pipeline = streamingPipeline;

    PCollection<Event> rawInput = pipeline.apply("Create Streaming Events", values);
    PCollection<KV<String, KV<Long, String>>> input =
        rawInput.apply("To KV", ParDo.of(new MapEventsToKV()));

    input = input.apply("Window input", Window.into(FixedWindows.of(Duration.standardSeconds(5))));

    StringBufferOrderedProcessingHandler handler =
        new StringBufferOrderedProcessingHandler(
            EMISSION_FREQUENCY_ON_EVERY_ELEMENT, INITIAL_SEQUENCE_OF_0);
    handler.setMaxOutputElementsPerBundle(LARGE_MAX_RESULTS_PER_OUTPUT);
    handler.setStatusUpdateFrequency(null);
    handler.setProduceStatusUpdateOnEveryEvent(true);

    OrderedEventProcessor<String, String, String, StringBuilderState> orderedEventProcessor =
        OrderedEventProcessor.create(handler);

    OrderedEventProcessorResult<String, String, String> processingResult =
        input.apply("Process Events", orderedEventProcessor);

    IntervalWindow window1 = new IntervalWindow(base, base.plus(Duration.standardSeconds(5)));
    PAssert.that("Output matches in window 1", processingResult.output())
        .inWindow(window1)
        .containsInAnyOrder(
            KV.of("id-1", "a"),
            KV.of("id-1", "ab"),
            KV.of("id-2", "x"),
            KV.of("id-2", "xy"),
            KV.of("id-2", "xyz"));

    IntervalWindow window2 =
        new IntervalWindow(
            base.plus(Duration.standardSeconds(10)), base.plus(Duration.standardSeconds(15)));
    PAssert.that("Output matches in window 2", processingResult.output())
        .inWindow(window2)
        .containsInAnyOrder(KV.of("id-1", "c"), KV.of("id-1", "cd"));

    PAssert.that("Statuses match in window 1", processingResult.processingStatuses())
        .inWindow(window1)
        .containsInAnyOrder(
            KV.of("id-1", OrderedProcessingStatus.create(0L, 0, null, null, 1, 1, 0, false)),
            KV.of("id-1", OrderedProcessingStatus.create(1L, 0, null, null, 2, 2, 0, false)),
            KV.of("id-2", OrderedProcessingStatus.create(0L, 0, null, null, 1, 1, 0, false)),
            KV.of("id-2", OrderedProcessingStatus.create(1L, 0, null, null, 2, 2, 0, false)),
            KV.of("id-2", OrderedProcessingStatus.create(2L, 0, null, null, 3, 3, 0, false)));

    PAssert.that("Statuses match in window 2", processingResult.processingStatuses())
        .inWindow(window2)
        .containsInAnyOrder(
            KV.of("id-1", OrderedProcessingStatus.create(0L, 0, null, null, 1, 1, 0, false)),
            KV.of("id-1", OrderedProcessingStatus.create(1L, 0, null, null, 2, 2, 0, false)));

    PAssert.that("Unprocessed events match", processingResult.unprocessedEvents())
        .containsInAnyOrder(NO_EXPECTED_DLQ_EVENTS);

    pipeline.run();
  }