[GitHub] [flink] 1996fanrui commented on a diff in pull request #20689: [FLINK-28984][runtime] Fix the problem that FsCheckpointStateOutputStream is not being released normally

[GitBox <gi...@apache.org>]

1996fanrui commented on code in PR #20689:
URL: https://github.com/apache/flink/pull/20689#discussion_r1048368938


##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java:
##########
@@ -454,6 +460,14 @@ private void createStream() throws IOException {
                                     fs, createStatePath(), WriteMode.NO_OVERWRITE);
                     this.outStream = streamAndPath.stream();
                     this.statePath = streamAndPath.path();
+                    if (closed) {
+                        cleanUp();
+                        LOG.info(
+                                "Closing this stream becaue it is closed! {}",

Review Comment:
   ```suggestion
                                   "Closing this stream because {} is closed.",
   ```



##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java:
##########
@@ -355,20 +356,25 @@ public void close() {
                 // that the stream is closed
                 pos = writeBuffer.length;
 
-                if (outStream != null) {

Review Comment:
   The `outStream` and `statePath` should be added the `volatile`.
   
   Async thread created the `outStream`, and then task thread closes the `FsCheckpointStateOutputStream`, and task thread should close the `outStream` and delete the `statePath`. However, the `outStream` and `statePath` didn't add the `volatile`, so task thread may not be able to read `outStream`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org