tkaymak commented on code in PR #40550:
URL: https://github.com/apache/beam/pull/40550#discussion_r4294267268
##########
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:
Please drop the late element, `LATE_INDEX`, the else branch in
`ExpiryKeySelectorFn` and this assertion. It pins the first visit behavior that
#40446 wants to tighten, and the zero row assertions already prove the fix,
without the change `beamWatermark` still has one row here. The window expires
anyway, the source watermark is declared on the source Dataset and advances on
the filtered elements too.
##########
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:
Contract only please, for example: `A key without timers and state keeps no
row, its next visit is a first visit.` The reasoning belongs in #40517, not in
the code.
##########
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:
`getAllowedTimestampSkew` is deprecated, with the late element gone this
override goes as well.
##########
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:
nit: `ExpiryTestFn` is `DownstreamFn` with another output string. Reuse
`DownstreamFn` and assert on `TestUnboundedSource.element(TAG_EXPIRY, 0, 0)`.
--
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]
