mardi 14 mai 2019

Apache Beam TestStream finalPane is empty

I'm in the process of writing a simple test to verify the semantics of early/on-time/late panes. The pipeline combines the number of elements per key. My early and on-time panes are working as expected, though my final pane seems to be empty at all times.

private static final Duration WINDOW_LENGTH = Duration.standardMinutes(2);
private static final Duration LATENESS_HORIZON = Duration.standardDays(1);

My test is as follows:

@Test
@Category(ValidatesRunner.class)
public void simpleTest() throws Exception {
    Instant baseTime = new Instant(0L);
    Duration one_min = Duration.standardMinutes(1);


    TestStream<KV<String, Long>> events = TestStream.create(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()))
            .advanceWatermarkTo(baseTime)

            // First element arrives
            .addElements(
                    TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
            )
            .advanceProcessingTime(Duration.standardMinutes(5))

            // Second element arrives
            .addElements(
                    TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
            )
            .advanceProcessingTime(Duration.standardMinutes(5))

            // Third element arrives
            .addElements(
                    TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
            )
            .advanceProcessingTime(Duration.standardMinutes(5))

            // Window ends
            .advanceWatermarkTo(baseTime.plus(WINDOW_LENGTH).plus(one_min))

            // Late element arrives
            .addElements(
                    TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
            )
            .advanceProcessingTime(Duration.standardMinutes(5))

            // Fire all
            .advanceWatermarkToInfinity();

    PCollection<KV<String, Long>> userCount = p.apply(events).apply(new CountPipeline());

    IntervalWindow window = new IntervalWindow(baseTime, WINDOW_LENGTH);

    PAssert.that(userCount)  // This test works
            .inEarlyPane(window)
            .containsInAnyOrder(
                KV.of("laurens", 1L),  // First firing
                KV.of("laurens", 2L),  // Second firing
                KV.of("laurens", 3L)   // Third firing
            );

    PAssert.that(userCount) // This test works as well
            .inOnTimePane(window)
            .containsInAnyOrder(
                    KV.of("laurens", 3L) // On time firing
            );

    PAssert.that(userCount) // Test fails
            .inFinalPane(window)
            .containsInAnyOrder(
                    KV.of("laurens", 4L) // Late firing
            );

    p.run().waitUntilFinish();
}

Code of the pipeline is as follows:

public static class CountPipeline extends PTransform<PCollection<KV<String, Long>>, PCollection<KV<String, Long>>> {

    @Override
    public PCollection<KV<String, Long>> expand(PCollection<KV<String, Long>> events) {
        return events.apply("window", Window.<KV<String, Long>>into(FixedWindows.of(WINDOW_LENGTH))
                        .triggering(AfterWatermark
                                .pastEndOfWindow()
                                .withEarlyFirings(AfterProcessingTime
                                        .pastFirstElementInPane())
                                .withLateFirings(AfterProcessingTime
                                        .pastFirstElementInPane())
                        )
                        .withAllowedLateness(LATENESS_HORIZON)
                        .accumulatingFiredPanes()
                        .withOnTimeBehavior(Window.OnTimeBehavior.FIRE_ALWAYS)
                ).apply("Count", Count.perKey());
    }
}

Error:

Expected: iterable over [<KV{laurens, 4}>] in any order
 but: No item matches: <KV{laurens, 4}> in []

As you can see, the last element is definitely ingested after the watermark, which should make it late by definition. Though, the final pane does not contain a refinement of the original result. I'm honestly clueless why it the late pane is not firing. Any insights are appreciated.

Aucun commentaire:

Enregistrer un commentaire