You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Zakelly (via GitHub)" <gi...@apache.org> on 2024/12/12 09:06:07 UTC
[PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Zakelly opened a new pull request, #25792:
URL: https://github.com/apache/flink/pull/25792
## What is the purpose of the change
The watermark status is not properly handle in async state processing. It is emitted to downstream synchronously, while the watermarks will respect the ongoing timer processing. That should be fixed and aligned.
## Brief change log
- Rewrite `processWatermarkStatus` in `AbstractAsyncStateStreamOperatorV2` and `AbstractAsyncStateStreamOperator`, making watermark status also respect the timers.
## Verifying this change
This change is already covered by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #25792:
URL: https://github.com/apache/flink/pull/25792#issuecomment-2544427450
Thanks for the review! Merging...
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25792:
URL: https://github.com/apache/flink/pull/25792#issuecomment-2538287976
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2c6e7e5107048c628a13f4520034f940a23f2fab",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2c6e7e5107048c628a13f4520034f940a23f2fab",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2c6e7e5107048c628a13f4520034f940a23f2fab UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
Yes your description of the logic makes sense.
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1881730342
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java:
##########
@@ -259,17 +265,35 @@ public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
return;
}
- asyncExecutionController.processNonRecord(
- () -> {
- // todo: make async operator deal with interruptible watermark
- if (timeServiceManager != null) {
- CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
- future.thenAccept(v -> output.emitWatermark(mark));
- asyncExecutionController.drainWithTimerIfNeeded(future);
- } else {
+ asyncExecutionController.processNonRecord(() -> doProcessWatermark(mark, null));
+ }
+
+ /**
+ * Handle the watermark and timers, then run a provided {@link Runnable} asynchronously right
+ * after the watermark is emitted.
+ *
+ * @param mark The watermark.
+ * @param postAction The runnable for post action.
+ */
+ protected void doProcessWatermark(Watermark mark, @Nullable Runnable postAction)
+ throws Exception {
+ // todo: make async operator deal with interruptible watermark
+ if (timeServiceManager != null) {
+ CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
+ future.thenAccept(
+ v -> {
output.emitWatermark(mark);
- }
- });
+ if (postAction != null) {
+ postAction.run();
+ }
+ });
+ asyncExecutionController.drainWithTimerIfNeeded(future);
+ } else {
+ output.emitWatermark(mark);
Review Comment:
nit: the future.thenAccept( and the else seem to have the same logic, can we put this in a function?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
Yes your description of the logic makes sense.
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
the parent class does these if independently with no `else`
```
protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index)
throws Exception {
boolean wasIdle = combinedWatermark.isIdle();
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
processWatermark(new Watermark(combinedWatermark.getCombinedWatermark()));
}
if (wasIdle != combinedWatermark.isIdle()) {
output.emitWatermarkStatus(watermarkStatus);
}
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883189632
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
This part is basically a rewrite of the overriden method, `AbstractStreamOperator#emitWatermarkStatus`, considering the async timer processing and emitting watermark or status after that. The logic in the method of parent class is more readable. Which is:
- If a combined watermark changes (because some of the watermarks become active), we process the combined watermark.
- If the combined watermark idle/active status switched, notify the downstream.
These two if(s) are independent.
This PR ensures watermark status emitting after the watermark processing under the async context. I don't think a comment or two would explain this logic well, but you can get the point if you compare this method with the one from parent class.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1881755602
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java:
##########
@@ -259,17 +265,35 @@ public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
return;
}
- asyncExecutionController.processNonRecord(
- () -> {
- // todo: make async operator deal with interruptible watermark
- if (timeServiceManager != null) {
- CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
- future.thenAccept(v -> output.emitWatermark(mark));
- asyncExecutionController.drainWithTimerIfNeeded(future);
- } else {
+ asyncExecutionController.processNonRecord(() -> doProcessWatermark(mark, null));
+ }
+
+ /**
+ * Handle the watermark and timers, then run a provided {@link Runnable} asynchronously right
+ * after the watermark is emitted.
+ *
+ * @param mark The watermark.
+ * @param postAction The runnable for post action.
+ */
+ protected void doProcessWatermark(Watermark mark, @Nullable Runnable postAction)
+ throws Exception {
+ // todo: make async operator deal with interruptible watermark
+ if (timeServiceManager != null) {
+ CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
+ future.thenAccept(
+ v -> {
output.emitWatermark(mark);
- }
- });
+ if (postAction != null) {
+ postAction.run();
+ }
+ });
+ asyncExecutionController.drainWithTimerIfNeeded(future);
+ } else {
+ output.emitWatermark(mark);
Review Comment:
Sure, done
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1882410030
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
it would be worth including a comment detailing what is happening here. Here is my understanding , please correct me if I am wrong:
I assume when the if is true: then the combined watermark status has changed, from either active to idle or the other way round. In this case we call doProcessWatermark, the first parameter news up a Watermark - I can only assume that we want to clone the combined watermark. If the idleness of the combined watermark is unchanged when we do nothing, otherwise we need to send an idle or active watermark , based on on the supplied watermarkStatus which should have the same state as the combined watermark.
In the else - if the combined watermark did not change then we check whether the existing combined watermark idleness is not the same as the supplied watermark. In this case why do we emit the watermark? I assume we could have 2 idle sources coming into an operator so the combined water mark would be idle, if one of the sources becomes active, then the combined watermark will not have changed status so why would we emit a watermark.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883809521
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
@davidradl `if`s are independent, but the execution order is serial. This PR must ensure in async context, the possible `processWatermark` and `output.emitWatermarkStatus` still executed in order. So if you read the code:
```
if (conditionA) {
asyncDo(a; then if (conditionB) then b)
} else if (conditionB) {
do(b)
}
```
The conditionB is split into two parts, that's why there is an `else`
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883187336
##########
flink-runtime/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java:
##########
@@ -51,20 +56,20 @@ public TwoInputStreamOperatorTestHarness(
super(operator, maxParallelism, numSubtasks, subtaskIndex);
this.twoInputOperator = operator;
+ processor1 = RecordProcessorUtils.getRecordProcessor1(twoInputOperator);
+ processor2 = RecordProcessorUtils.getRecordProcessor2(twoInputOperator);
}
public void processElement1(StreamRecord<IN1> element) throws Exception {
- twoInputOperator.setKeyContextElement1(element);
- twoInputOperator.processElement1(element);
+ processor1.accept(element);
Review Comment:
It looks like `twoInputOperator.setKeyContextElement1(element);` is still needed, otherwise, the test will fail.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1884125298
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Ah I see; and if we removed the else effectively we would be doing the synchronous processing incorrectly for the async case. Thanks for the clarification
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
Yes your description of the logic makes sense.
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
the parent class does these ifs independently with no `else`
```
protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index)
throws Exception {
boolean wasIdle = combinedWatermark.isIdle();
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
processWatermark(new Watermark(combinedWatermark.getCombinedWatermark()));
}
if (wasIdle != combinedWatermark.isIdle()) {
output.emitWatermarkStatus(watermarkStatus);
}
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1882419555
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java:
##########
@@ -259,17 +265,33 @@ public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
return;
}
- asyncExecutionController.processNonRecord(
- () -> {
- // todo: make async operator deal with interruptible watermark
- if (timeServiceManager != null) {
- CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
- future.thenAccept(v -> output.emitWatermark(mark));
- asyncExecutionController.drainWithTimerIfNeeded(future);
- } else {
- output.emitWatermark(mark);
- }
- });
+ asyncExecutionController.processNonRecord(() -> doProcessWatermark(mark, null));
+ }
+
+ /**
+ * Handle the watermark and timers, then run a provided {@link Runnable} asynchronously right
+ * after the watermark is emitted.
+ *
+ * @param mark The watermark.
+ * @param postAction The runnable for post action.
+ */
+ protected void doProcessWatermark(Watermark mark, @Nullable Runnable postAction)
+ throws Exception {
+ // todo: make async operator deal with interruptible watermark
Review Comment:
I am curious what is an interruptible watermark?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
you say the 2 `if`s are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`s are independent?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883199135
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperatorV2.java:
##########
@@ -259,17 +265,33 @@ public void processWatermark(Watermark mark) throws Exception {
super.processWatermark(mark);
return;
}
- asyncExecutionController.processNonRecord(
- () -> {
- // todo: make async operator deal with interruptible watermark
- if (timeServiceManager != null) {
- CompletableFuture<Void> future = timeServiceManager.advanceWatermark(mark);
- future.thenAccept(v -> output.emitWatermark(mark));
- asyncExecutionController.drainWithTimerIfNeeded(future);
- } else {
- output.emitWatermark(mark);
- }
- });
+ asyncExecutionController.processNonRecord(() -> doProcessWatermark(mark, null));
+ }
+
+ /**
+ * Handle the watermark and timers, then run a provided {@link Runnable} asynchronously right
+ * after the watermark is emitted.
+ *
+ * @param mark The watermark.
+ * @param postAction The runnable for post action.
+ */
+ protected void doProcessWatermark(Watermark mark, @Nullable Runnable postAction)
+ throws Exception {
+ // todo: make async operator deal with interruptible watermark
Review Comment:
Please read https://issues.apache.org/jira/browse/FLINK-20217
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly merged PR #25792:
URL: https://github.com/apache/flink/pull/25792
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
Yes your description of the logic makes sense.
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
the parent class does this if independantly with no else
` protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index)
throws Exception {
boolean wasIdle = combinedWatermark.isIdle();
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
processWatermark(new Watermark(combinedWatermark.getCombinedWatermark()));
}
if (wasIdle != combinedWatermark.isIdle()) {
output.emitWatermarkStatus(watermarkStatus);
}`
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883693470
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
Thanks @Zakelly
Yes your description of the logic makes sense.
you say the 2 `if` 's are independent, but the second `if `is in an `else`. Shouldn't we remove the `else` so the `if`'s are independent?
the parent class does this if independantly with no else
```
protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index)
throws Exception {
boolean wasIdle = combinedWatermark.isIdle();
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
processWatermark(new Watermark(combinedWatermark.getCombinedWatermark()));
}
if (wasIdle != combinedWatermark.isIdle()) {
output.emitWatermarkStatus(watermarkStatus);
}
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883809521
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
@davidradl `if`s are independent, but the execution order is serial. This PR must ensure in async context, the possible `processWatermark` and `output.emitWatermarkStatus` still executed in order. So if you read the code:
```
if (conditionA) {
asyncDo(a; then if (conditionB) then b)
} else if (conditionB) {
do(b)
}
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #25792:
URL: https://github.com/apache/flink/pull/25792#issuecomment-2541732285
Reviewed by Chi on 12/12/24. appears that this PR is healthily progressing
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1882410030
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
it would be worth including a comment detailing what is happening here. Here is my understanding , please correct me if I am wrong:
I assume when the if is true: then the combined watermark status has changed, from either active to idle or the other way round. In this case we call doProcessWatermark, the first parameter news up a Watermark - I can only assume that we want to clone the combined watermark. If the idleness of the combined watermark is unchanged when we do nothing, otherwise we need to send an idle or active watermark , based on on the supplied watermarkStatus which should have the same status as the combined watermark.
In the else - if the combined watermark did not change then we check whether the existing combined watermark idleness is not the same as the supplied watermark. In this case why do we emit the watermark? I assume we could have 2 idle sources coming into an operator so the combined water mark would be idle, if one of the sources becomes active, then the combined watermark will not have changed status so why would we emit a watermark.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1882410030
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
it would be worth including a comment detailing what is happening here. Here is my understanding , please correct me if I am wrong:
I assume when the if is true: then the combined watermark status has changed, from either active to idle or the other way round. In this case we call doProcessWatermark, the first parameter news up a Watermark - I can only assume that we want to clone the combined watermark. If the idleness of the combined watermark is unchanged when we do nothing, otherwise we need to send an idle or active watermark , based on on the supplied watermarkStatus which should have the same status as the combined watermark.
In the else - if the combined watermark did not change then we check whether the existing combined watermark idleness is not the same as the supplied watermark.
- I assume we could have 2 idle sources coming into an operator so the combined watermark status would be idle, if one of the sources becomes active, then the combined watermark will now be able to progress ignoring the other idle source.
- I assume we could have 2 active sources coming into an operator so the combined watermark status would be active, if one of the sources becomes idle, then the combined watermark will still be able to progress ignoring the idle source. In this case why do we need to send a watermark?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883417520
##########
flink-runtime/src/test/java/org/apache/flink/streaming/util/TwoInputStreamOperatorTestHarness.java:
##########
@@ -51,20 +56,20 @@ public TwoInputStreamOperatorTestHarness(
super(operator, maxParallelism, numSubtasks, subtaskIndex);
this.twoInputOperator = operator;
+ processor1 = RecordProcessorUtils.getRecordProcessor1(twoInputOperator);
+ processor2 = RecordProcessorUtils.getRecordProcessor2(twoInputOperator);
}
public void processElement1(StreamRecord<IN1> element) throws Exception {
- twoInputOperator.setKeyContextElement1(element);
- twoInputOperator.processElement1(element);
+ processor1.accept(element);
Review Comment:
I reverted the last commit, since every test is not properly consider & handle the threads relation.
We will have a new operator test harness for async state processing.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36892][Runtime] Properly handle the watermark status within async state processing [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #25792:
URL: https://github.com/apache/flink/pull/25792#discussion_r1883809521
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncStateStreamOperator.java:
##########
@@ -368,10 +385,12 @@ protected void processWatermarkStatus(WatermarkStatus watermarkStatus, int index
boolean wasIdle = combinedWatermark.isIdle();
// index is 0-based
if (combinedWatermark.updateStatus(index, watermarkStatus.isIdle())) {
- super.processWatermark(
- new Watermark(combinedWatermark.getCombinedWatermark()));
- }
- if (wasIdle != combinedWatermark.isIdle()) {
+ doProcessWatermark(
Review Comment:
@davidradl `if`s are independent, but the execution order is serial. This PR must ensure in async context, the possible `processWatermark` and `output.emitWatermarkStatus` still executed in order. So if you read the code:
```
if (conditionA) {
asyncDo(a; then if (conditionB) then b)
} else if (conditionB) {
do(b)
}
```
The conditionB and actionB are split from the latter `if`, that's why there is an `else`
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org