pnowojski commented on code in PR #25167:
URL: https://github.com/apache/flink/pull/25167#discussion_r1715152842
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java:
##########
@@ -273,6 +322,16 @@ void emitPeriodicWatermark() {
}
watermarkMultiplexer.onPeriodicEmit();
}
+
+ public void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String>
splitsToResume) {
+ for (String splitId : splitsToPause) {
+ inputActivityClocks.get(splitId).markBlocked();
+ }
+ for (String splitId : splitsToResume) {
+ inputActivityClocks.get(splitId).markUnblocked();
+ }
Review Comment:
But that RPC contains only `long maxWatermark` value. So yes, that value
might be outdated due to race conditions. But splits are purely handled in the
source operator's thread, which should take care of internal consistency of
it's fields.
In other words, when RPC with `WatermarkAlignmentEvent` is received, it will
be processed in the source operator's thread. `SourceOperator` will then
iterate over it's currently active (not closed/released) splits in
`checkSplitWatermarkAlignment()` method, and call `pauseOrResumeSplits()`
appropriately. On the other hand, outputs/splits are released also from
`SourceOperator`'s thread, during
`org.apache.flink.streaming.api.operators.SourceOperator#emitNext`. So there
should be no opportunity for a race condition between those two.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
