[
https://issues.apache.org/jira/browse/FLINK-41445?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Weiqing Yang updated FLINK-41445:
---------------------------------
Description:
*What happens*
{\{RestoreTestBase#enablePerRecordWatermarks}} and
\{{CommonSemanticTestBase#createSourceOptions}} switch the values source to
per-record watermarks (\{{scan.watermark.emit.strategy}} = \{{on-event}}) when
the source declares a watermark. The comment in \{{RestoreTestBase}} explains
why: processing-time watermark emission would otherwise lead to flaky tests.
Both detect the watermark with a case-sensitive check:
{code:java}
sourceTestStep.schemaComponents.stream().anyMatch(c -> c.startsWith("WATERMARK
FOR"))
{code}
23 schema components in the exec node test programs write \{{WATERMARK for}}
(18) or \{{watermark for}} (5). Those sources never get per-record watermarks
and silently run with periodic watermark emission instead.
*Affected test classes* (each runs at least one program with a lower-case
spelling)
* Restore tests:
** \{{DeduplicationRestoreTest}}
** \{{GroupWindowAggregateEventTimeRestoreTest}}
** \{{GroupWindowAggregateProcTimeRestoreTest}}
** \{{IntervalJoinRestoreTest}}
** \{{IntervalJoinProcTimeEarlyFireRestoreTest}}
** \{{MatchRecognizeRestoreTest}}
** \{{MiniBatchAssignerRestoreTest}}
** \{{OverAggregateRestoreTest}}
** \{{OverWindowRestoreTest}}
** \{{TableSourceScanRestoreTest}}
** \{{TemporalSortRestoreTest}}
** \{{WatermarkAssignerRestoreTest}}
** \{{WindowAggregateEventTimeRestoreTest}}
** \{{WindowDeduplicateEventTimeRestoreTest}}
** \{{WindowJoinEventTimeRestoreTest}}
** \{{WindowRankRestoreTest}}
** \{{WindowTableFunctionEventTimeRestoreTest}}
** \{{WindowTableFunctionProcTimeRestoreTest}}
* Semantic tests:
** \{{WindowJoinSemanticTests}}
** \{{MiscSemanticTests}}
The batch test bases are not affected; they do not set the watermark emission
strategy.
*Proposed fix*
Compare case-insensitively in both places:
{code:java}
c.toUpperCase(Locale.ROOT).startsWith("WATERMARK FOR")
{code}
With this change, all 20 classes above pass against the committed plans and
savepoints (143 tests, 0 failures), so no test files need to be regenerated.
*Relation to FLINK-40967*
FLINK-40967 assumes the generator for
\{{over-aggregate-bounded-non-partitioned-rows-with-out-of-order-records}} runs
with per-record watermarks. Its source
(\{{OverAggregateTestPrograms#getSourceTestStep}}) writes \{{watermark for
rowtime as rowtime}}, so on master it does not. With the fix above applied,
regenerating that program still produces no savepoint within 5 minutes, while
\{{over-aggregate-bounded-partitioned-rows-with-out-of-order-records}}
regenerates normally. So FLINK-40967 is a separate problem and is not fixed by
this change.
was:
{\{RestoreTestBase#enablePerRecordWatermarks}} and
\{{CommonSemanticTestBase#createSourceOptions}} switch the values source to
per-record watermarks
(\{{scan.watermark.emit.strategy = on-event}}) whenever the source declares a
watermark. The comment in \{{RestoreTestBase}} gives the reason: processing-time
watermark emission would otherwise lead to flaky tests.
\{code:java}
sourceTestStep.schemaComponents.stream().anyMatch(c ->
c.startsWith("WATERMARK FOR"))
\{code}
23 schema components in the exec node test programs write \{{WATERMARK for}}
(18) or \{{watermark for}} (5). Those sources never get per-record watermarks
and
silently run with periodic watermark emission instead.
Test classes with at least one affected program:
* Restore tests: \{{DeduplicationRestoreTest}},
\{{GroupWindowAggregateEventTimeRestoreTest}},
\{{GroupWindowAggregateProcTimeRestoreTest}},
\{{IntervalJoinRestoreTest}}, \{{IntervalJoinProcTimeEarlyFireRestoreTest}},
\{{MatchRecognizeRestoreTest}}, \{{MiniBatchAssignerRestoreTest}},
\{{OverAggregateRestoreTest}}, \{{OverWindowRestoreTest}},
\{{TableSourceScanRestoreTest}}, \{{TemporalSortRestoreTest}},
\{{WatermarkAssignerRestoreTest}},
\{{WindowAggregateEventTimeRestoreTest}},
\{{WindowDeduplicateEventTimeRestoreTest}},
\{{WindowJoinEventTimeRestoreTest}}, \{{WindowRankRestoreTest}},
\{{WindowTableFunctionEventTimeRestoreTest}},
\{{WindowTableFunctionProcTimeRestoreTest}}
* Semantic tests: \{{WindowJoinSemanticTests}}, \{{MiscSemanticTests}}
The batch test bases are not affected; they do not set the watermark emission
strategy.
*Proposed fix:* compare case-insensitively in both places
(\{{c.toUpperCase(Locale.ROOT).startsWith("WATERMARK FOR")}}). With that
change, all 20 classes above pass
against the committed plans and savepoints (143 tests, 0 failures), so no
test files need to be regenerated.
*Relation to FLINK-40967:* that ticket assumes the generator for
\{{over-aggregate-bounded-non-partitioned-rows-with-out-of-order-records}} runs
with per-record
watermarks. Its source (\{{OverAggregateTestPrograms#getSourceTestStep}})
writes \{{watermark for rowtime as rowtime}}, so on master it does not. With
the fix above
applied, regenerating that program still produces no savepoint within 5
minutes, while
\{{over-aggregate-bounded-partitioned-rows-with-out-of-order-records}}
regenerates normally. So FLINK-40967 is a separate problem and is not fixed
by this change.
> Restore and semantic tests skip per-record watermarks when a test schema
> writes "WATERMARK for" in lower case
> --------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41445
> URL: https://issues.apache.org/jira/browse/FLINK-41445
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner, Tests
> Affects Versions: 2.4.0
> Reporter: Weiqing Yang
> Assignee: Weiqing Yang
> Priority: Minor
>
> *What happens*
> {\{RestoreTestBase#enablePerRecordWatermarks}} and
> \{{CommonSemanticTestBase#createSourceOptions}} switch the values source to
> per-record watermarks (\{{scan.watermark.emit.strategy}} = \{{on-event}})
> when the source declares a watermark. The comment in \{{RestoreTestBase}}
> explains why: processing-time watermark emission would otherwise lead to
> flaky tests.
> Both detect the watermark with a case-sensitive check:
> {code:java}
> sourceTestStep.schemaComponents.stream().anyMatch(c ->
> c.startsWith("WATERMARK FOR"))
> {code}
> 23 schema components in the exec node test programs write \{{WATERMARK for}}
> (18) or \{{watermark for}} (5). Those sources never get per-record watermarks
> and silently run with periodic watermark emission instead.
> *Affected test classes* (each runs at least one program with a lower-case
> spelling)
> * Restore tests:
> ** \{{DeduplicationRestoreTest}}
> ** \{{GroupWindowAggregateEventTimeRestoreTest}}
> ** \{{GroupWindowAggregateProcTimeRestoreTest}}
> ** \{{IntervalJoinRestoreTest}}
> ** \{{IntervalJoinProcTimeEarlyFireRestoreTest}}
> ** \{{MatchRecognizeRestoreTest}}
> ** \{{MiniBatchAssignerRestoreTest}}
> ** \{{OverAggregateRestoreTest}}
> ** \{{OverWindowRestoreTest}}
> ** \{{TableSourceScanRestoreTest}}
> ** \{{TemporalSortRestoreTest}}
> ** \{{WatermarkAssignerRestoreTest}}
> ** \{{WindowAggregateEventTimeRestoreTest}}
> ** \{{WindowDeduplicateEventTimeRestoreTest}}
> ** \{{WindowJoinEventTimeRestoreTest}}
> ** \{{WindowRankRestoreTest}}
> ** \{{WindowTableFunctionEventTimeRestoreTest}}
> ** \{{WindowTableFunctionProcTimeRestoreTest}}
> * Semantic tests:
> ** \{{WindowJoinSemanticTests}}
> ** \{{MiscSemanticTests}}
> The batch test bases are not affected; they do not set the watermark emission
> strategy.
> *Proposed fix*
> Compare case-insensitively in both places:
> {code:java}
> c.toUpperCase(Locale.ROOT).startsWith("WATERMARK FOR")
> {code}
> With this change, all 20 classes above pass against the committed plans and
> savepoints (143 tests, 0 failures), so no test files need to be regenerated.
> *Relation to FLINK-40967*
> FLINK-40967 assumes the generator for
> \{{over-aggregate-bounded-non-partitioned-rows-with-out-of-order-records}}
> runs with per-record watermarks. Its source
> (\{{OverAggregateTestPrograms#getSourceTestStep}}) writes \{{watermark for
> rowtime as rowtime}}, so on master it does not. With the fix above applied,
> regenerating that program still produces no savepoint within 5 minutes, while
> \{{over-aggregate-bounded-partitioned-rows-with-out-of-order-records}}
> regenerates normally. So FLINK-40967 is a separate problem and is not fixed
> by this change.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
