[jira] [Created] (FLINK-40500) DataStream V2 idleness detection is not backpressure-aware - FLIP-471 was never applied to ExtractEventTimeProcessFunction

["Martijn Visser (Jira)" <ji...@apache.org>]

Martijn Visser created FLINK-40500:
--------------------------------------

             Summary: DataStream V2 idleness detection is not backpressure-aware - FLIP-471 was never applied to ExtractEventTimeProcessFunction
                 Key: FLINK-40500
                 URL: https://issues.apache.org/jira/browse/FLINK-40500
             Project: Flink
          Issue Type: Bug
          Components: API / DataStream
            Reporter: Martijn Visser


FLINK-35886 (FLIP-471) fixed incorrect idleness-timeout accounting when a subtask is
backpressured or blocked by watermark alignment, by introducing
{{PausableRelativeClock}}. Every DataStream V1 path uses it
({{ProgressiveTimestampsAndWatermarks}}, {{TimestampsAndWatermarksOperator}},
{{SourceOperator}}).

The DataStream V2 {{ExtractEventTimeProcessFunction}} reuses
{{WatermarksWithIdleness.IdlenessTimer}} but constructs it with the raw
{{processingTimeService.getClock()}} ({{ExtractEventTimeProcessFunction.java:90-92}});
the hosting operators pass the service through unwrapped ({{ProcessOperator.java:115-120}}
and siblings). There is no {{PausableRelativeClock}} and no backpressure-listener
registration anywhere in flink-datastream (verified by grep). The V2 code postdates the
FLIP-471 fix by five months, so this is a missed carry-over, not a merge-ordering issue.

Consequence: under sustained backpressure, a V2 pipeline with an idle timeout declares
inputs idle while records are queued; the combined watermark advances past them and they
are dropped as late — exactly the failure FLIP-471 fixed for V1.

Characterization test:
{{ExtractEventTimeProcessFunctionTest#testIdleStatusEmittedPurelyOnWallClockElapse}} —
with idleTimeout=200ms, idle=true is emitted purely on wall-clock elapse; no mechanism
exists by which runtime-induced blocking could suppress it.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)