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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
