dawidwys commented on a change in pull request #15550:
URL: https://github.com/apache/flink/pull/15550#discussion_r610715281
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##########
@@ -932,9 +932,9 @@ public void heartbeatFromResourceManager(ResourceID
resourceID) {
final Task task = taskSlotTable.getTask(executionAttemptID);
if (task != null) {
- task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp,
checkpointOptions);
-
- return CompletableFuture.completedFuture(Acknowledge.get());
+ return task.triggerCheckpointBarrier(
+ checkpointId, checkpointTimestamp,
checkpointOptions)
+ .thenApply(result -> Acknowledge.get());
Review comment:
Right, I am also not sure about the change. However if we go with
ignoring the result (option 2. from the description) we should revisit the
whole stack, as there is a somewhat bigger logic in
`org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask#pendingCheckpointCompletedFutures`
which does not work now.
Is it fine to drop it? Especially cancelling futures for the oldest pending
checkpoints? If it is fine to drop (it does not work anyway) then I am also +1
to drop the return value whatsoever.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##########
@@ -932,9 +932,9 @@ public void heartbeatFromResourceManager(ResourceID
resourceID) {
final Task task = taskSlotTable.getTask(executionAttemptID);
if (task != null) {
- task.triggerCheckpointBarrier(checkpointId, checkpointTimestamp,
checkpointOptions);
-
- return CompletableFuture.completedFuture(Acknowledge.get());
+ return task.triggerCheckpointBarrier(
+ checkpointId, checkpointTimestamp,
checkpointOptions)
+ .thenApply(result -> Acknowledge.get());
Review comment:
Right, I am also not sure about the change. However if we go with
ignoring the result (option 2. from the description) we should revisit the
whole stack, as there is a somewhat bigger logic in
`org.apache.flink.streaming.runtime.tasks.MultipleInputStreamTask#pendingCheckpointCompletedFutures`
which does not work now. Is it fine to drop it? Especially cancelling futures
for the oldest pending checkpoints? If it is fine to drop (it does not work
anyway) then I am also +1 to drop the return value whatsoever.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
