[FLINK-36125] Fix concurrent duplicated discard of logical file in cp…

… file-merging (apache#25236)

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/LogicalFile.java

@@ -25,6 +25,7 @@
 
 import java.io.IOException;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.apache.flink.runtime.checkpoint.filemerging.FileMergingSnapshotManager.SubtaskKey;
 
@@ -59,7 +60,7 @@ public static LogicalFileId generateRandomId() {
     private long lastUsedCheckpointID = -1L;
 
     /** Whether this logical file is removed by checkpoint subsumption/abortion. */
-    boolean discarded = false;
+    AtomicBoolean discarded = new AtomicBoolean(false);
 
     /** The physical file where this logical file is stored. This should never be null. */
     @Nonnull private final PhysicalFile physicalFile;
@@ -114,10 +115,9 @@ public void advanceLastCheckpointId(long checkpointId) {
      * @throws IOException if anything goes wrong with file system.
      */
     public void discardWithCheckpointId(long checkpointId) throws IOException {
-        if (!discarded && checkpointId >= lastUsedCheckpointID) {
+        if (checkpointId >= lastUsedCheckpointID && discarded.compareAndSet(false, true)) {
             physicalFile.decRefCount();
             physicalFile.decSize(length);
-            discarded = true;
         }
     }
 
@@ -145,7 +145,7 @@ public SubtaskKey getSubtaskKey() {
 
     @VisibleForTesting
     public boolean isDiscarded() {
-        return discarded;
+        return discarded.get();
     }
 
     @Override

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerTestBase.java

@@ -180,6 +180,10 @@ void testRefCountBetweenLogicalAndPhysicalFiles() throws IOException {
             assertThat(physicalFile1.isDeleted()).isFalse();
             assertThat(physicalFile1.getRefCount()).isZero();
 
+            // duplicated discard takes no effect
+            logicalFile1.discardWithCheckpointId(2);
+            assertThat(physicalFile1.getRefCount()).isZero();
+
             physicalFile1.close();
             assertThat(physicalFile1.isOpen()).isFalse();
             assertThat(physicalFile1.isDeleted()).isTrue();