You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2021/02/25 10:31:22 UTC
[flink] 01/03: [hotfix][task] Rename isStoppingBySyncSavepoint to
ignoreEndOfInput
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 017b33db14bebbc956a035c7c10dd2605152bbe1
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Feb 23 21:00:37 2021 +0100
[hotfix][task] Rename isStoppingBySyncSavepoint to ignoreEndOfInput
---
.../org/apache/flink/streaming/api/operators/StreamSource.java | 2 +-
.../apache/flink/streaming/runtime/tasks/OperatorChain.java | 10 +++++-----
.../apache/flink/streaming/runtime/tasks/SourceStreamTask.java | 2 +-
.../org/apache/flink/streaming/runtime/tasks/StreamTask.java | 6 +++---
4 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index 30ade24..8ab8a6b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -119,7 +119,7 @@ public class StreamSource<OUT, SRC extends SourceFunction<OUT>>
// interface,
// so we still need the following call to end the input
synchronized (lockingObject) {
- operatorChain.setIsStoppingBySyncSavepoint(false);
+ operatorChain.setIgnoreEndOfInput(false);
operatorChain.endHeadOperatorInput(1);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
index 9af6320..74dfbf5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
@@ -105,7 +105,7 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>> implements Strea
private final OperatorEventDispatcherImpl operatorEventDispatcher;
- private boolean isStoppingBySyncSavepoint;
+ private boolean ignoreEndOfInput;
/**
* Current status of the input stream of the operator chain. Watermarks explicitly generated by
@@ -291,7 +291,7 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>> implements Strea
* @param inputId the input ID starts from 1 which indicates the first input.
*/
public void endHeadOperatorInput(int inputId) throws Exception {
- if (headOperatorWrapper != null && !isStoppingBySyncSavepoint) {
+ if (headOperatorWrapper != null && !ignoreEndOfInput) {
headOperatorWrapper.endOperatorInput(inputId);
}
}
@@ -317,7 +317,7 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>> implements Strea
*/
protected void closeOperators(StreamTaskActionExecutor actionExecutor) throws Exception {
if (headOperatorWrapper != null) {
- headOperatorWrapper.close(actionExecutor, isStoppingBySyncSavepoint);
+ headOperatorWrapper.close(actionExecutor, ignoreEndOfInput);
}
}
@@ -582,8 +582,8 @@ public class OperatorChain<OUT, OP extends StreamOperator<OUT>> implements Strea
return (tailOperatorWrapper == null) ? null : tailOperatorWrapper.getStreamOperator();
}
- public void setIsStoppingBySyncSavepoint(boolean stoppingBySyncSavepoint) {
- this.isStoppingBySyncSavepoint = stoppingBySyncSavepoint;
+ public void setIgnoreEndOfInput(boolean ignoreEndOfInput) {
+ this.ignoreEndOfInput = ignoreEndOfInput;
}
// ------------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index e749c52..db1fba9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -253,7 +253,7 @@ public class SourceStreamTask<
headOperator.run(lock, getStreamStatusMaintainer(), operatorChain);
if (!wasStoppedExternally && !isCanceled()) {
synchronized (lock) {
- operatorChain.setIsStoppingBySyncSavepoint(false);
+ operatorChain.setIgnoreEndOfInput(false);
}
}
completionFuture.complete(null);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 4d7b3c7..32b8612 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -404,7 +404,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
if (!succeeded && activeSyncSavepointId != null && activeSyncSavepointId == id) {
// allow to process further EndOfPartition events
activeSyncSavepointId = null;
- operatorChain.setIsStoppingBySyncSavepoint(false);
+ operatorChain.setIgnoreEndOfInput(false);
}
syncSavepointId = null;
}
@@ -415,7 +415,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
"at most one stop-with-savepoint checkpoint at a time is allowed");
syncSavepointId = checkpointId;
activeSyncSavepointId = checkpointId;
- operatorChain.setIsStoppingBySyncSavepoint(true);
+ operatorChain.setIgnoreEndOfInput(true);
}
@VisibleForTesting
@@ -999,7 +999,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> extends Ab
} else if (activeSyncSavepointId != null
&& activeSyncSavepointId < checkpointMetaData.getCheckpointId()) {
activeSyncSavepointId = null;
- operatorChain.setIsStoppingBySyncSavepoint(false);
+ operatorChain.setIgnoreEndOfInput(false);
}
subtaskCheckpointCoordinator.checkpointState(