(1.19 backport) [FLINK-35886][task] Fix watermark idleness timeout accounting when subtask is backpressured/blocked #25223
Conversation
…t pausing splits in MockSourceReader
Operators have to be serializable and they will also need an access to Clock to construct ProgressBlockingRelativeClock. Because we also want to be able to provide for testing purposes ManualClock we have to find a way how Operators could obtain a Clock instance. Exposing Clock from ProcessingTimeService sounds like a good place as it also will provide a since source of processing time for potential users (for example ProgressBlockingRelativeClock and firing timers)
…stampsAndWatermarksOperator
…leness detection in SourceOperator
…rmarkAssignerOperator
| <!-- New method has been added to the interface, but this interface shouldn't be implemented by users, only used/called, so this doesn't break API compatibility. --> | ||
| <exclude>org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier$Context</exclude> | ||
| <!-- Base interface has been extracted from the existing Clock class. This interface shouldn't affect user code. --> | ||
| <exclude>org.apache.flink.util.clock.Clock</exclude> |
There was a problem hiding this comment.
Please pay attention to this
There was a problem hiding this comment.
It would be nice to return System clock from the new method (WatermarkGeneratorSupplier.Context#getInputActivityClock) to avoid breaking existing implementations.
I think it's not impossible that there are custom implementations of this interface.
There was a problem hiding this comment.
This context is provided by Flink to operators/functions, so I don't see any legitimate way how user should be interested in implementing it's own context. This could only happen if someone forks Flink and extends it somehow in this area AFAIU. I will merge it as is.
There was a problem hiding this comment.
Actually this WatermarkGeneratorSupplier.Context#getInputActivityClock, turns out to be a problem and user code actually implements on its own WatermarkGeneratorSupplier.Context interface. For example our Kafka connector does this here: https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java#L438-L439
Due to this reason, I don't see an easy way how to backport this bug fix to release 1.19 and 1.20 and I would suggest to only fix the bug in 2.0. Especially given that there doesn't seem to be a large interest in this bug fix (it hasn't been reported before). I will revert my changes for 1.19 and 1.20
| <!-- New method has been added to the interface, but this interface shouldn't be implemented by users, only used/called, so this doesn't break API compatibility. --> | ||
| <exclude>org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier$Context</exclude> | ||
| <!-- Base interface has been extracted from the existing Clock class. This interface shouldn't affect user code. --> | ||
| <exclude>org.apache.flink.util.clock.Clock</exclude> |
There was a problem hiding this comment.
It would be nice to return System clock from the new method (WatermarkGeneratorSupplier.Context#getInputActivityClock) to avoid breaking existing implementations.
I think it's not impossible that there are custom implementations of this interface.
this is a backport of #25167