You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/03/11 18:00:09 UTC
[GitHub] [flink] rkhachatryan opened a new pull request #11382:
[FLINK-16554][task] split StreamTask
rkhachatryan opened a new pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382
## What is the purpose of the change
StreamTask is currently 1400+ LOC.
We can cut it to 1100+ by simply extracting these static classes into separate files:
`CheckpointingOperation`
`AsyncCheckpointRunnable`
NOTE: The scope of the refactoring is intentionally small because it's a sensitive part of the system.
## Brief change log
- *extract `CheckpointingOperation` and `AsyncCheckpointRunnable` into separate files*
- *don't pass StreamTask as a parameter*
## Verifying this change
This change is a trivial rework without any test coverage.
## 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/Mesos, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153145930",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 3b898115d9c7e949cc6751b911098e9b2ee0f295 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153145930) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] rkhachatryan commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392086726
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+final class AsyncCheckpointRunnable implements Runnable, Closeable {
+
+ private final String taskName;
+ private final CloseableRegistry closeableRegistry;
+ private final Environment taskEnvironment;
+
+ private enum AsyncCheckpointState {
+ RUNNING,
+ DISCARDED,
+ COMPLETED
+ }
+
+ private final AsyncExceptionHandler asyncExceptionHandler;
+ private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
+ private final CheckpointMetaData checkpointMetaData;
+ private final CheckpointMetrics checkpointMetrics;
+ private final long asyncStartNanos;
+ private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
+
+ AsyncCheckpointRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
+ CheckpointMetaData checkpointMetaData,
+ CheckpointMetrics checkpointMetrics,
+ long asyncStartNanos,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ Environment taskEnvironment,
+ AsyncExceptionHandler asyncExceptionHandler) {
+
+ this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
+ this.checkpointMetaData = checkNotNull(checkpointMetaData);
+ this.checkpointMetrics = checkNotNull(checkpointMetrics);
+ this.asyncStartNanos = asyncStartNanos;
+ this.taskName = checkNotNull(taskName);
+ this.closeableRegistry = checkNotNull(closeableRegistry);
+ this.taskEnvironment = checkNotNull(taskEnvironment);
+ this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ }
+
+ @Override
+ public void run() {
+ FileSystemSafetyNet.initializeSafetyNetForThread();
+ try {
+ closeableRegistry.registerCloseable(this);
+
+ TaskStateSnapshot jobManagerTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ TaskStateSnapshot localTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+
+ for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry : operatorSnapshotsInProgress.entrySet()) {
+
+ OperatorID operatorID = entry.getKey();
+ OperatorSnapshotFutures snapshotInProgress = entry.getValue();
+
+ // finalize the async part of all by executing all snapshot runnables
+ OperatorSnapshotFinalizer finalizedSnapshots =
+ new OperatorSnapshotFinalizer(snapshotInProgress);
+
+ jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getJobManagerOwnedState());
+
+ localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getTaskLocalState());
+ }
+
+ final long asyncEndNanos = System.nanoTime();
+ final long asyncDurationMillis = (asyncEndNanos - asyncStartNanos) / 1_000_000L;
+
+ checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
+
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
+
+ reportCompletedSnapshotStates(
+ jobManagerTaskOperatorSubtaskStates,
+ localTaskOperatorSubtaskStates,
+ asyncDurationMillis);
+
+ } else {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
+ taskName,
+ checkpointMetaData.getCheckpointId());
+ }
+ } catch (Exception e) {
+ if (StreamTask.LOG.isDebugEnabled()) {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed.",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ e);
+ }
+ handleExecutionException(e);
+ } finally {
+ closeableRegistry.unregisterCloseable(this);
+ FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
+ }
+ }
+
+ private void reportCompletedSnapshotStates(
+ TaskStateSnapshot acknowledgedTaskStateSnapshot,
+ TaskStateSnapshot localTaskStateSnapshot,
+ long asyncDurationMillis) {
+
+ boolean hasAckState = acknowledgedTaskStateSnapshot.hasState();
+ boolean hasLocalState = localTaskStateSnapshot.hasState();
+
+ Preconditions.checkState(hasAckState || !hasLocalState,
+ "Found cached state but no corresponding primary state is reported to the job " +
+ "manager. This indicates a problem.");
+
+ // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
+ // to stateless tasks on restore. This enables simple job modifications that only concern
+ // stateless without the need to assign them uids to match their (always empty) states.
+ taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
+ checkpointMetaData,
+ checkpointMetrics,
+ hasAckState ? acknowledgedTaskStateSnapshot : null,
+ hasLocalState ? localTaskStateSnapshot : null);
+
+ StreamTask.LOG.debug("{} - finished asynchronous part of checkpoint {}. Asynchronous duration: {} ms",
+ taskName, checkpointMetaData.getCheckpointId(), asyncDurationMillis);
+
+ StreamTask.LOG.trace("{} - reported the following states in snapshot for checkpoint {}: {}.",
+ taskName, checkpointMetaData.getCheckpointId(), acknowledgedTaskStateSnapshot);
+ }
+
+ private void handleExecutionException(Exception e) {
+
+ boolean didCleanup = false;
+ AsyncCheckpointState currentState = asyncCheckpointState.get();
+
+ while (AsyncCheckpointState.DISCARDED != currentState) {
+
+ if (asyncCheckpointState.compareAndSet(currentState, AsyncCheckpointState.DISCARDED)) {
+
+ didCleanup = true;
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+
+ Exception checkpointException = new Exception(
+ "Could not materialize checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " +
+ taskName + '.',
+ e);
+
+ // We only report the exception for the original cause of fail and cleanup.
+ // Otherwise this followup exception could race the original exception in failing the task.
+ try {
+ taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(), checkpointException);
+ } catch (Exception unhandled) {
+ AsynchronousException asyncException = new AsynchronousException(unhandled);
+ asyncExceptionHandler.handleAsyncException("Failure in asynchronous checkpoint materialization", asyncException);
+ }
+
+ currentState = AsyncCheckpointState.DISCARDED;
+ } else {
+ currentState = asyncCheckpointState.get();
+ }
+ }
+
+ if (!didCleanup) {
+ StreamTask.LOG.trace("Caught followup exception from a failed checkpoint thread. This can be ignored.", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.DISCARDED)) {
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ StreamTask.LOG.warn("Could not properly clean up the async checkpoint runnable.", cleanupException);
+ }
+ } else {
+ logFailedCleanupAttempt();
+ }
+ }
+
+ private void cleanup() throws Exception {
+ StreamTask.LOG.debug(
Review comment:
It's already exposed ~~for the other classes~~ but yes, it's better not to use it.
I'll change it to a private logger.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/152865648) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196)
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153145930",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153122310) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264)
* 3b898115d9c7e949cc6751b911098e9b2ee0f295 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153145930) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/152865648) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196)
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/152865648) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196)
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153122310) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW merged pull request #11382: [FLINK-16554][task]
split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW merged pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153122310) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392026565
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/CheckpointingOperation.java
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.concurrent.ExecutorService;
+
+final class CheckpointingOperation {
+
+ static void execute(
+ CheckpointMetaData checkpointMetaData,
+ CheckpointOptions checkpointOptions,
+ CheckpointMetrics checkpointMetrics,
+ CheckpointStreamFactory storageLocation,
+ OperatorChain<?, ?> operatorChain,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ ExecutorService threadPool,
+ Environment environment,
+ AsyncExceptionHandler asyncExceptionHandler) throws Exception {
+
+ Preconditions.checkNotNull(checkpointMetaData);
+ Preconditions.checkNotNull(checkpointOptions);
+ Preconditions.checkNotNull(checkpointMetrics);
+ Preconditions.checkNotNull(storageLocation);
Review comment:
Also checkNotNull for other arguments?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392026596
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -928,14 +919,16 @@ private void checkpointState(
checkpointMetaData.getCheckpointId(),
checkpointOptions.getTargetLocation());
- CheckpointingOperation checkpointingOperation = new CheckpointingOperation(
- this,
- checkpointMetaData,
+ CheckpointingOperation.execute(checkpointMetaData,
Review comment:
nit: also make `checkpointMetaData` in separate line?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/152865648) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196)
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11382: [FLINK-16554][task]
split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] TisonKun commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
TisonKun commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392039386
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+final class AsyncCheckpointRunnable implements Runnable, Closeable {
+
+ private final String taskName;
+ private final CloseableRegistry closeableRegistry;
+ private final Environment taskEnvironment;
+
+ private enum AsyncCheckpointState {
+ RUNNING,
+ DISCARDED,
+ COMPLETED
+ }
+
+ private final AsyncExceptionHandler asyncExceptionHandler;
+ private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
+ private final CheckpointMetaData checkpointMetaData;
+ private final CheckpointMetrics checkpointMetrics;
+ private final long asyncStartNanos;
+ private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
+
+ AsyncCheckpointRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
+ CheckpointMetaData checkpointMetaData,
+ CheckpointMetrics checkpointMetrics,
+ long asyncStartNanos,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ Environment taskEnvironment,
+ AsyncExceptionHandler asyncExceptionHandler) {
+
+ this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
+ this.checkpointMetaData = checkNotNull(checkpointMetaData);
+ this.checkpointMetrics = checkNotNull(checkpointMetrics);
+ this.asyncStartNanos = asyncStartNanos;
+ this.taskName = checkNotNull(taskName);
+ this.closeableRegistry = checkNotNull(closeableRegistry);
+ this.taskEnvironment = checkNotNull(taskEnvironment);
+ this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ }
+
+ @Override
+ public void run() {
+ FileSystemSafetyNet.initializeSafetyNetForThread();
+ try {
+ closeableRegistry.registerCloseable(this);
+
+ TaskStateSnapshot jobManagerTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ TaskStateSnapshot localTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+
+ for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry : operatorSnapshotsInProgress.entrySet()) {
+
+ OperatorID operatorID = entry.getKey();
+ OperatorSnapshotFutures snapshotInProgress = entry.getValue();
+
+ // finalize the async part of all by executing all snapshot runnables
+ OperatorSnapshotFinalizer finalizedSnapshots =
+ new OperatorSnapshotFinalizer(snapshotInProgress);
+
+ jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getJobManagerOwnedState());
+
+ localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getTaskLocalState());
+ }
+
+ final long asyncEndNanos = System.nanoTime();
+ final long asyncDurationMillis = (asyncEndNanos - asyncStartNanos) / 1_000_000L;
+
+ checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
+
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
+
+ reportCompletedSnapshotStates(
+ jobManagerTaskOperatorSubtaskStates,
+ localTaskOperatorSubtaskStates,
+ asyncDurationMillis);
+
+ } else {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
+ taskName,
+ checkpointMetaData.getCheckpointId());
+ }
+ } catch (Exception e) {
+ if (StreamTask.LOG.isDebugEnabled()) {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed.",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ e);
+ }
+ handleExecutionException(e);
+ } finally {
+ closeableRegistry.unregisterCloseable(this);
+ FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
+ }
+ }
+
+ private void reportCompletedSnapshotStates(
+ TaskStateSnapshot acknowledgedTaskStateSnapshot,
+ TaskStateSnapshot localTaskStateSnapshot,
+ long asyncDurationMillis) {
+
+ boolean hasAckState = acknowledgedTaskStateSnapshot.hasState();
+ boolean hasLocalState = localTaskStateSnapshot.hasState();
+
+ Preconditions.checkState(hasAckState || !hasLocalState,
+ "Found cached state but no corresponding primary state is reported to the job " +
+ "manager. This indicates a problem.");
+
+ // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
+ // to stateless tasks on restore. This enables simple job modifications that only concern
+ // stateless without the need to assign them uids to match their (always empty) states.
+ taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
+ checkpointMetaData,
+ checkpointMetrics,
+ hasAckState ? acknowledgedTaskStateSnapshot : null,
+ hasLocalState ? localTaskStateSnapshot : null);
+
+ StreamTask.LOG.debug("{} - finished asynchronous part of checkpoint {}. Asynchronous duration: {} ms",
+ taskName, checkpointMetaData.getCheckpointId(), asyncDurationMillis);
+
+ StreamTask.LOG.trace("{} - reported the following states in snapshot for checkpoint {}: {}.",
+ taskName, checkpointMetaData.getCheckpointId(), acknowledgedTaskStateSnapshot);
+ }
+
+ private void handleExecutionException(Exception e) {
+
+ boolean didCleanup = false;
+ AsyncCheckpointState currentState = asyncCheckpointState.get();
+
+ while (AsyncCheckpointState.DISCARDED != currentState) {
+
+ if (asyncCheckpointState.compareAndSet(currentState, AsyncCheckpointState.DISCARDED)) {
+
+ didCleanup = true;
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+
+ Exception checkpointException = new Exception(
+ "Could not materialize checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " +
+ taskName + '.',
+ e);
+
+ // We only report the exception for the original cause of fail and cleanup.
+ // Otherwise this followup exception could race the original exception in failing the task.
+ try {
+ taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(), checkpointException);
+ } catch (Exception unhandled) {
+ AsynchronousException asyncException = new AsynchronousException(unhandled);
+ asyncExceptionHandler.handleAsyncException("Failure in asynchronous checkpoint materialization", asyncException);
+ }
+
+ currentState = AsyncCheckpointState.DISCARDED;
+ } else {
+ currentState = asyncCheckpointState.get();
+ }
+ }
+
+ if (!didCleanup) {
+ StreamTask.LOG.trace("Caught followup exception from a failed checkpoint thread. This can be ignored.", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.DISCARDED)) {
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ StreamTask.LOG.warn("Could not properly clean up the async checkpoint runnable.", cleanupException);
+ }
+ } else {
+ logFailedCleanupAttempt();
+ }
+ }
+
+ private void cleanup() throws Exception {
+ StreamTask.LOG.debug(
Review comment:
Shall we pass the logger from outside instead of expose it as `public/(default)`?
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] rkhachatryan commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392086726
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+final class AsyncCheckpointRunnable implements Runnable, Closeable {
+
+ private final String taskName;
+ private final CloseableRegistry closeableRegistry;
+ private final Environment taskEnvironment;
+
+ private enum AsyncCheckpointState {
+ RUNNING,
+ DISCARDED,
+ COMPLETED
+ }
+
+ private final AsyncExceptionHandler asyncExceptionHandler;
+ private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
+ private final CheckpointMetaData checkpointMetaData;
+ private final CheckpointMetrics checkpointMetrics;
+ private final long asyncStartNanos;
+ private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
+
+ AsyncCheckpointRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
+ CheckpointMetaData checkpointMetaData,
+ CheckpointMetrics checkpointMetrics,
+ long asyncStartNanos,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ Environment taskEnvironment,
+ AsyncExceptionHandler asyncExceptionHandler) {
+
+ this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
+ this.checkpointMetaData = checkNotNull(checkpointMetaData);
+ this.checkpointMetrics = checkNotNull(checkpointMetrics);
+ this.asyncStartNanos = asyncStartNanos;
+ this.taskName = checkNotNull(taskName);
+ this.closeableRegistry = checkNotNull(closeableRegistry);
+ this.taskEnvironment = checkNotNull(taskEnvironment);
+ this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ }
+
+ @Override
+ public void run() {
+ FileSystemSafetyNet.initializeSafetyNetForThread();
+ try {
+ closeableRegistry.registerCloseable(this);
+
+ TaskStateSnapshot jobManagerTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ TaskStateSnapshot localTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+
+ for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry : operatorSnapshotsInProgress.entrySet()) {
+
+ OperatorID operatorID = entry.getKey();
+ OperatorSnapshotFutures snapshotInProgress = entry.getValue();
+
+ // finalize the async part of all by executing all snapshot runnables
+ OperatorSnapshotFinalizer finalizedSnapshots =
+ new OperatorSnapshotFinalizer(snapshotInProgress);
+
+ jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getJobManagerOwnedState());
+
+ localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getTaskLocalState());
+ }
+
+ final long asyncEndNanos = System.nanoTime();
+ final long asyncDurationMillis = (asyncEndNanos - asyncStartNanos) / 1_000_000L;
+
+ checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
+
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
+
+ reportCompletedSnapshotStates(
+ jobManagerTaskOperatorSubtaskStates,
+ localTaskOperatorSubtaskStates,
+ asyncDurationMillis);
+
+ } else {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
+ taskName,
+ checkpointMetaData.getCheckpointId());
+ }
+ } catch (Exception e) {
+ if (StreamTask.LOG.isDebugEnabled()) {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed.",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ e);
+ }
+ handleExecutionException(e);
+ } finally {
+ closeableRegistry.unregisterCloseable(this);
+ FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
+ }
+ }
+
+ private void reportCompletedSnapshotStates(
+ TaskStateSnapshot acknowledgedTaskStateSnapshot,
+ TaskStateSnapshot localTaskStateSnapshot,
+ long asyncDurationMillis) {
+
+ boolean hasAckState = acknowledgedTaskStateSnapshot.hasState();
+ boolean hasLocalState = localTaskStateSnapshot.hasState();
+
+ Preconditions.checkState(hasAckState || !hasLocalState,
+ "Found cached state but no corresponding primary state is reported to the job " +
+ "manager. This indicates a problem.");
+
+ // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
+ // to stateless tasks on restore. This enables simple job modifications that only concern
+ // stateless without the need to assign them uids to match their (always empty) states.
+ taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
+ checkpointMetaData,
+ checkpointMetrics,
+ hasAckState ? acknowledgedTaskStateSnapshot : null,
+ hasLocalState ? localTaskStateSnapshot : null);
+
+ StreamTask.LOG.debug("{} - finished asynchronous part of checkpoint {}. Asynchronous duration: {} ms",
+ taskName, checkpointMetaData.getCheckpointId(), asyncDurationMillis);
+
+ StreamTask.LOG.trace("{} - reported the following states in snapshot for checkpoint {}: {}.",
+ taskName, checkpointMetaData.getCheckpointId(), acknowledgedTaskStateSnapshot);
+ }
+
+ private void handleExecutionException(Exception e) {
+
+ boolean didCleanup = false;
+ AsyncCheckpointState currentState = asyncCheckpointState.get();
+
+ while (AsyncCheckpointState.DISCARDED != currentState) {
+
+ if (asyncCheckpointState.compareAndSet(currentState, AsyncCheckpointState.DISCARDED)) {
+
+ didCleanup = true;
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+
+ Exception checkpointException = new Exception(
+ "Could not materialize checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " +
+ taskName + '.',
+ e);
+
+ // We only report the exception for the original cause of fail and cleanup.
+ // Otherwise this followup exception could race the original exception in failing the task.
+ try {
+ taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(), checkpointException);
+ } catch (Exception unhandled) {
+ AsynchronousException asyncException = new AsynchronousException(unhandled);
+ asyncExceptionHandler.handleAsyncException("Failure in asynchronous checkpoint materialization", asyncException);
+ }
+
+ currentState = AsyncCheckpointState.DISCARDED;
+ } else {
+ currentState = asyncCheckpointState.get();
+ }
+ }
+
+ if (!didCleanup) {
+ StreamTask.LOG.trace("Caught followup exception from a failed checkpoint thread. This can be ignored.", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.DISCARDED)) {
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ StreamTask.LOG.warn("Could not properly clean up the async checkpoint runnable.", cleanupException);
+ }
+ } else {
+ logFailedCleanupAttempt();
+ }
+ }
+
+ private void cleanup() throws Exception {
+ StreamTask.LOG.debug(
Review comment:
It's already exposed ``for the other classes`` but yes, it's better not to use it.
I'll change it to a private logger.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392026854
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/LocalStateForwardingTest.java
##########
@@ -105,13 +105,12 @@ public void testReportingFromSnapshotToTaskStateManager() {
OperatorID operatorID = new OperatorID();
snapshots.put(operatorID, osFuture);
- StreamTask.AsyncCheckpointRunnable checkpointRunnable =
- new StreamTask.AsyncCheckpointRunnable(
- testStreamTask,
+ AsyncCheckpointRunnable checkpointRunnable =
+ new AsyncCheckpointRunnable(
snapshots,
checkpointMetaData,
checkpointMetrics,
- 0L);
+ 0L, testStreamTask.getName(), testStreamTask.getCancelables(), testStreamTask.getEnvironment(), testStreamTask);
Review comment:
nit: split line for every field.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153145930",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 3b898115d9c7e949cc6751b911098e9b2ee0f295 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153145930) Azure: [SUCCESS](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6278)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392026468
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+@VisibleForTesting
+final class AsyncCheckpointRunnable implements Runnable, Closeable {
+
+ private final String taskName;
+ private final CloseableRegistry closeableRegistry;
+ private final Environment taskEnvironment;
+
+ private enum AsyncCheckpointState {
+ RUNNING,
+ DISCARDED,
+ COMPLETED
+ }
+
+ private final AsyncExceptionHandler asyncExceptionHandler;
+ private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
+ private final CheckpointMetaData checkpointMetaData;
+ private final CheckpointMetrics checkpointMetrics;
+ private final long asyncStartNanos;
+ private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
+
+ AsyncCheckpointRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
+ CheckpointMetaData checkpointMetaData,
+ CheckpointMetrics checkpointMetrics,
+ long asyncStartNanos,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ Environment taskEnvironment,
+ AsyncExceptionHandler asyncExceptionHandler) {
+
+ this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
+ this.checkpointMetaData = checkNotNull(checkpointMetaData);
+ this.checkpointMetrics = checkNotNull(checkpointMetrics);
+ this.asyncStartNanos = asyncStartNanos;
+ this.taskName = checkNotNull(taskName);
+ this.closeableRegistry = checkNotNull(closeableRegistry);
+ this.taskEnvironment = checkNotNull(taskEnvironment);
+ this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ }
+
+ @Override
+ public void run() {
+ FileSystemSafetyNet.initializeSafetyNetForThread();
+ try {
+ closeableRegistry.registerCloseable(this);
+
+ TaskStateSnapshot jobManagerTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ TaskStateSnapshot localTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+
+ for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry : operatorSnapshotsInProgress.entrySet()) {
+
+ OperatorID operatorID = entry.getKey();
+ OperatorSnapshotFutures snapshotInProgress = entry.getValue();
+
+ // finalize the async part of all by executing all snapshot runnables
+ OperatorSnapshotFinalizer finalizedSnapshots =
+ new OperatorSnapshotFinalizer(snapshotInProgress);
+
+ jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getJobManagerOwnedState());
+
+ localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getTaskLocalState());
+ }
+
+ final long asyncEndNanos = System.nanoTime();
+ final long asyncDurationMillis = (asyncEndNanos - asyncStartNanos) / 1_000_000L;
+
+ checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
+
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
+
+ reportCompletedSnapshotStates(
+ jobManagerTaskOperatorSubtaskStates,
+ localTaskOperatorSubtaskStates,
+ asyncDurationMillis);
+
+ } else {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
+ taskName,
+ checkpointMetaData.getCheckpointId());
+ }
+ } catch (Exception e) {
+ if (StreamTask.LOG.isDebugEnabled()) {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed.",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ e);
+ }
+ handleExecutionException(e);
+ } finally {
+ closeableRegistry.unregisterCloseable(this);
+ FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
+ }
+ }
+
+ private void reportCompletedSnapshotStates(
+ TaskStateSnapshot acknowledgedTaskStateSnapshot,
+ TaskStateSnapshot localTaskStateSnapshot,
+ long asyncDurationMillis) {
+
+ boolean hasAckState = acknowledgedTaskStateSnapshot.hasState();
+ boolean hasLocalState = localTaskStateSnapshot.hasState();
+
+ Preconditions.checkState(hasAckState || !hasLocalState,
+ "Found cached state but no corresponding primary state is reported to the job " +
+ "manager. This indicates a problem.");
+
+ // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
+ // to stateless tasks on restore. This enables simple job modifications that only concern
+ // stateless without the need to assign them uids to match their (always empty) states.
+ this.taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
Review comment:
nit: remove `this` for consistency.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] rkhachatryan commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392086726
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+final class AsyncCheckpointRunnable implements Runnable, Closeable {
+
+ private final String taskName;
+ private final CloseableRegistry closeableRegistry;
+ private final Environment taskEnvironment;
+
+ private enum AsyncCheckpointState {
+ RUNNING,
+ DISCARDED,
+ COMPLETED
+ }
+
+ private final AsyncExceptionHandler asyncExceptionHandler;
+ private final Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress;
+ private final CheckpointMetaData checkpointMetaData;
+ private final CheckpointMetrics checkpointMetrics;
+ private final long asyncStartNanos;
+ private final AtomicReference<AsyncCheckpointState> asyncCheckpointState = new AtomicReference<>(AsyncCheckpointState.RUNNING);
+
+ AsyncCheckpointRunnable(
+ Map<OperatorID, OperatorSnapshotFutures> operatorSnapshotsInProgress,
+ CheckpointMetaData checkpointMetaData,
+ CheckpointMetrics checkpointMetrics,
+ long asyncStartNanos,
+ String taskName,
+ CloseableRegistry closeableRegistry,
+ Environment taskEnvironment,
+ AsyncExceptionHandler asyncExceptionHandler) {
+
+ this.operatorSnapshotsInProgress = checkNotNull(operatorSnapshotsInProgress);
+ this.checkpointMetaData = checkNotNull(checkpointMetaData);
+ this.checkpointMetrics = checkNotNull(checkpointMetrics);
+ this.asyncStartNanos = asyncStartNanos;
+ this.taskName = checkNotNull(taskName);
+ this.closeableRegistry = checkNotNull(closeableRegistry);
+ this.taskEnvironment = checkNotNull(taskEnvironment);
+ this.asyncExceptionHandler = checkNotNull(asyncExceptionHandler);
+ }
+
+ @Override
+ public void run() {
+ FileSystemSafetyNet.initializeSafetyNetForThread();
+ try {
+ closeableRegistry.registerCloseable(this);
+
+ TaskStateSnapshot jobManagerTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+ TaskStateSnapshot localTaskOperatorSubtaskStates = new TaskStateSnapshot(operatorSnapshotsInProgress.size());
+
+ for (Map.Entry<OperatorID, OperatorSnapshotFutures> entry : operatorSnapshotsInProgress.entrySet()) {
+
+ OperatorID operatorID = entry.getKey();
+ OperatorSnapshotFutures snapshotInProgress = entry.getValue();
+
+ // finalize the async part of all by executing all snapshot runnables
+ OperatorSnapshotFinalizer finalizedSnapshots =
+ new OperatorSnapshotFinalizer(snapshotInProgress);
+
+ jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getJobManagerOwnedState());
+
+ localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
+ operatorID,
+ finalizedSnapshots.getTaskLocalState());
+ }
+
+ final long asyncEndNanos = System.nanoTime();
+ final long asyncDurationMillis = (asyncEndNanos - asyncStartNanos) / 1_000_000L;
+
+ checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);
+
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.COMPLETED)) {
+
+ reportCompletedSnapshotStates(
+ jobManagerTaskOperatorSubtaskStates,
+ localTaskOperatorSubtaskStates,
+ asyncDurationMillis);
+
+ } else {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",
+ taskName,
+ checkpointMetaData.getCheckpointId());
+ }
+ } catch (Exception e) {
+ if (StreamTask.LOG.isDebugEnabled()) {
+ StreamTask.LOG.debug("{} - asynchronous part of checkpoint {} could not be completed.",
+ taskName,
+ checkpointMetaData.getCheckpointId(),
+ e);
+ }
+ handleExecutionException(e);
+ } finally {
+ closeableRegistry.unregisterCloseable(this);
+ FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();
+ }
+ }
+
+ private void reportCompletedSnapshotStates(
+ TaskStateSnapshot acknowledgedTaskStateSnapshot,
+ TaskStateSnapshot localTaskStateSnapshot,
+ long asyncDurationMillis) {
+
+ boolean hasAckState = acknowledgedTaskStateSnapshot.hasState();
+ boolean hasLocalState = localTaskStateSnapshot.hasState();
+
+ Preconditions.checkState(hasAckState || !hasLocalState,
+ "Found cached state but no corresponding primary state is reported to the job " +
+ "manager. This indicates a problem.");
+
+ // we signal stateless tasks by reporting null, so that there are no attempts to assign empty state
+ // to stateless tasks on restore. This enables simple job modifications that only concern
+ // stateless without the need to assign them uids to match their (always empty) states.
+ taskEnvironment.getTaskStateManager().reportTaskStateSnapshots(
+ checkpointMetaData,
+ checkpointMetrics,
+ hasAckState ? acknowledgedTaskStateSnapshot : null,
+ hasLocalState ? localTaskStateSnapshot : null);
+
+ StreamTask.LOG.debug("{} - finished asynchronous part of checkpoint {}. Asynchronous duration: {} ms",
+ taskName, checkpointMetaData.getCheckpointId(), asyncDurationMillis);
+
+ StreamTask.LOG.trace("{} - reported the following states in snapshot for checkpoint {}: {}.",
+ taskName, checkpointMetaData.getCheckpointId(), acknowledgedTaskStateSnapshot);
+ }
+
+ private void handleExecutionException(Exception e) {
+
+ boolean didCleanup = false;
+ AsyncCheckpointState currentState = asyncCheckpointState.get();
+
+ while (AsyncCheckpointState.DISCARDED != currentState) {
+
+ if (asyncCheckpointState.compareAndSet(currentState, AsyncCheckpointState.DISCARDED)) {
+
+ didCleanup = true;
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+
+ Exception checkpointException = new Exception(
+ "Could not materialize checkpoint " + checkpointMetaData.getCheckpointId() + " for operator " +
+ taskName + '.',
+ e);
+
+ // We only report the exception for the original cause of fail and cleanup.
+ // Otherwise this followup exception could race the original exception in failing the task.
+ try {
+ taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(), checkpointException);
+ } catch (Exception unhandled) {
+ AsynchronousException asyncException = new AsynchronousException(unhandled);
+ asyncExceptionHandler.handleAsyncException("Failure in asynchronous checkpoint materialization", asyncException);
+ }
+
+ currentState = AsyncCheckpointState.DISCARDED;
+ } else {
+ currentState = asyncCheckpointState.get();
+ }
+ }
+
+ if (!didCleanup) {
+ StreamTask.LOG.trace("Caught followup exception from a failed checkpoint thread. This can be ignored.", e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (asyncCheckpointState.compareAndSet(AsyncCheckpointState.RUNNING, AsyncCheckpointState.DISCARDED)) {
+
+ try {
+ cleanup();
+ } catch (Exception cleanupException) {
+ StreamTask.LOG.warn("Could not properly clean up the async checkpoint runnable.", cleanupException);
+ }
+ } else {
+ logFailedCleanupAttempt();
+ }
+ }
+
+ private void cleanup() throws Exception {
+ StreamTask.LOG.debug(
Review comment:
It's already exposed for the other classes but yes, it's better not to use it.
I'll change it to a private logger.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11382: [FLINK-16554][task]
split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597783374
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit eb546e4bbf07492886da99bfac28496af894cd8c (Wed Mar 11 18:02:47 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-16554).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#discussion_r392026418
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.tasks;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.core.fs.CloseableRegistry;
+import org.apache.flink.core.fs.FileSystemSafetyNet;
+import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFinalizer;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * This runnable executes the asynchronous parts of all involved backend snapshots for the subtask.
+ */
+@VisibleForTesting
Review comment:
nit: I guess we do not need this annotation. It can not be defined as private even if no tests.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "SUCCESS",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3b898115d9c7e949cc6751b911098e9b2ee0f295",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153122310) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264)
* 3b898115d9c7e949cc6751b911098e9b2ee0f295 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11382:
[FLINK-16554][task] split StreamTask
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-597788287
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/152865648",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196",
"triggerID" : "eb546e4bbf07492886da99bfac28496af894cd8c",
"triggerType" : "PUSH"
}, {
"hash" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "55525e050ada84dcd0e1ea935dffe3b7f09faa42",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/153122310",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
}, {
"hash" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264",
"triggerID" : "823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* eb546e4bbf07492886da99bfac28496af894cd8c Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/152865648) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6196)
* 55525e050ada84dcd0e1ea935dffe3b7f09faa42 UNKNOWN
* 823cf79209f3ee0b6a38f9a185cdd5e506f3c6ea Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153122310) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6264)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] rkhachatryan commented on issue #11382: [FLINK-16554][task]
split StreamTask
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on issue #11382: [FLINK-16554][task] split StreamTask
URL: https://github.com/apache/flink/pull/11382#issuecomment-598549360
Thanks for reviewing @zhijiangW .
I've addressed the issues and squashed commits.
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services