[
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:
*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.
> 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)
