schizophrenicmaniac commented on code in PR #40550:
URL: https://github.com/apache/beam/pull/40550#discussion_r4296457473
##########
runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java:
##########
@@ -167,6 +172,44 @@ public void testChainedStatefulParDoTimerOutputSurvives()
throws Exception {
assertTrue(collected.contains("timer-output"));
}
+ @Test
+ public void testExpiredKeyLeavesNoState() throws Exception {
+ String checkpoint = tempFolder.newFolder("cp").getAbsolutePath();
+ String collectorId = StreamingTestUtils.newCollectorId(TAG_EXPIRY);
+ SparkStructuredStreamingPipelineOptions options =
+ StreamingTestUtils.streamingOptions(checkpoint);
+ options.setMaxRecordsPerBatch(1L);
+ Pipeline pipeline = Pipeline.create(options);
+ pipeline
+ .apply("Read", Read.from(new TestUnboundedSource(TAG_EXPIRY, 1, 8)))
+ .apply("KeySelector", ParDo.of(new ExpiryKeySelectorFn()))
+ .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
+ .apply("Window", Window.into(FixedWindows.of(Duration.millis(1000))))
+ .apply("StatefulParDo", ParDo.of(new ExpiryTestFn()))
+ .apply("Collect", ParDo.of(new
StreamingTestUtils.CollectDoFn<>(collectorId)));
+
+ assertEquals(PipelineResult.State.DONE,
StreamingTestUtils.run(pipeline).getState());
+ Set<String> collected = StreamingTestUtils.collected(collectorId);
+ assertTrue(collected.contains("k1-0"));
+ // The late element finds the key cleared and is filtered like a first
visit, see #40446.
+ assertTrue(collected.contains("k1-" + LATE_INDEX));
Review Comment:
Done, the late element, LATE_INDEX, the else branch and the assertion are
gone. The test now only checks the element got through and that the three state
variables have no rows.
##########
runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java:
##########
@@ -288,6 +331,38 @@ public void onTimer(@Timestamp Instant ts,
OutputReceiver<KV<String, String>> ou
}
}
+ private static final class ExpiryKeySelectorFn extends DoFn<String,
KV<String, String>> {
+ @ProcessElement
+ public void process(@Element String element, OutputReceiver<KV<String,
String>> out) {
+ int i = TestUnboundedSource.indexOf(element);
+ if (i == 0) {
+ out.output(KV.of("k1", element));
+ } else if (i == LATE_INDEX) {
+ // Back into the first window of k1, expired by now.
+ out.outputWithTimestamp(KV.of("k1", element), new
Instant(TestUnboundedSource.BASE_MILLIS));
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
Review Comment:
Removed.
##########
runners/spark/4/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/state/BeamStatefulProcessor.java:
##########
@@ -119,13 +119,23 @@ public void init(OutputMode outputMode, TimeMode
timeMode) {
}
private void storeWatermark(SparkTimerInternals timerInternals) {
-
checkStateNotNull(beamWatermark).update(timerInternals.currentInputWatermarkTime().getMillis());
+ ValueState<Long> watermarkState = checkStateNotNull(beamWatermark);
+ // A key without timers and state, e.g. after all its windows expired,
keeps nothing in the
Review Comment:
Done, I used your wording.
##########
runners/spark/4/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/StatefulParDoStreamingTest.java:
##########
@@ -288,6 +331,38 @@ public void onTimer(@Timestamp Instant ts,
OutputReceiver<KV<String, String>> ou
}
}
+ private static final class ExpiryKeySelectorFn extends DoFn<String,
KV<String, String>> {
+ @ProcessElement
+ public void process(@Element String element, OutputReceiver<KV<String,
String>> out) {
+ int i = TestUnboundedSource.indexOf(element);
+ if (i == 0) {
+ out.output(KV.of("k1", element));
+ } else if (i == LATE_INDEX) {
+ // Back into the first window of k1, expired by now.
+ out.outputWithTimestamp(KV.of("k1", element), new
Instant(TestUnboundedSource.BASE_MILLIS));
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return Duration.millis(Long.MAX_VALUE);
+ }
+ }
+
+ private static final class ExpiryTestFn extends DoFn<KV<String, String>,
String> {
Review Comment:
Done, switched to DownstreamFn and the assertion is on
TestUnboundedSource.element(TAG_EXPIRY, 0, 0) now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
