[FLINK-40555][runtime] Isolate unreadable JobResultStore entries per job

flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java

@@ -20,6 +20,7 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.ClusterOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.HighAvailabilityOptions;
 import org.apache.flink.core.fs.FileStatus;
@@ -47,6 +48,7 @@
 import java.io.OutputStream;
 import java.util.HashSet;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Executor;
 
@@ -63,6 +65,12 @@ public class FileSystemJobResultStore extends AbstractThreadsafeJobResultStore {
     @VisibleForTesting static final String FILE_EXTENSION = ".json";
     @VisibleForTesting static final String DIRTY_FILE_EXTENSION = "_DIRTY" + FILE_EXTENSION;
 
+    /**
+     * Suffix appended to a dirty result file that could not be read/parsed so that it is moved out
+     * of the way and no longer matches {@link #hasValidDirtyJobResultStoreEntryExtension(String)}.
+     */
+    @VisibleForTesting static final String QUARANTINED_FILE_SUFFIX = ".corrupted";
+
     @VisibleForTesting
     public static boolean hasValidDirtyJobResultStoreEntryExtension(String filename) {
         return filename.endsWith(DIRTY_FILE_EXTENSION);
@@ -83,13 +91,26 @@ public static boolean hasValidJobResultStoreEntryExtension(String filename) {
 
     private final boolean deleteOnCommit;
 
+    private final boolean jobErrorIsolationEnabled;
+
     @VisibleForTesting
     FileSystemJobResultStore(
             FileSystem fileSystem, Path basePath, boolean deleteOnCommit, Executor ioExecutor) {
+        this(fileSystem, basePath, deleteOnCommit, ioExecutor, false);
+    }
+
+    @VisibleForTesting
+    FileSystemJobResultStore(
+            FileSystem fileSystem,
+            Path basePath,
+            boolean deleteOnCommit,
+            Executor ioExecutor,
+            boolean jobErrorIsolationEnabled) {
         super(ioExecutor);
         this.fileSystem = fileSystem;
         this.basePath = basePath;
         this.deleteOnCommit = deleteOnCommit;
+        this.jobErrorIsolationEnabled = jobErrorIsolationEnabled;
     }
 
     public static FileSystemJobResultStore fromConfiguration(
@@ -109,7 +130,11 @@ public static FileSystemJobResultStore fromConfiguration(
         boolean deleteOnCommit = config.get(JobResultStoreOptions.DELETE_ON_COMMIT);
 
         return new FileSystemJobResultStore(
-                basePath.getFileSystem(), basePath, deleteOnCommit, ioExecutor);
+                basePath.getFileSystem(),
+                basePath,
+                deleteOnCommit,
+                ioExecutor,
+                config.get(ClusterOptions.JOB_ERROR_ISOLATION_ENABLED));
     }
 
     private void createBasePathIfNeeded() throws IOException {
@@ -208,16 +233,65 @@ public Set<JobResult> getDirtyResultsInternal() throws IOException {
         for (FileStatus s : statuses) {
             if (!s.isDir()) {
                 if (hasValidDirtyJobResultStoreEntryExtension(s.getPath().getName())) {
-                    JsonJobResultEntry jre =
-                            mapper.readValue(
-                                    fileSystem.open(s.getPath()), JsonJobResultEntry.class);
-                    dirtyResults.add(jre.getJobResult());
+                    readDirtyResult(s.getPath()).ifPresent(dirtyResults::add);
                 }
             }
         }
         return dirtyResults;
     }
 
+    /**
+     * Reads and parses a single dirty result file. If it cannot be read/parsed and {@link
+     * ClusterOptions#JOB_ERROR_ISOLATION_ENABLED} is enabled, the file is quarantined and {@link
+     * Optional#empty()} is returned instead of propagating the failure: neither its terminal status
+     * nor its application can be recovered from an unreadable file, and this codebase requires
+     * every dirty {@link JobResult} to have both, so no placeholder is synthesized for it.
+     */
+    private Optional<JobResult> readDirtyResult(Path path) throws IOException {
+        try {
+            final JsonJobResultEntry jre =
+                    mapper.readValue(fileSystem.open(path), JsonJobResultEntry.class);
+            return Optional.of(jre.getJobResult());
+        } catch (IOException e) {
+            if (!jobErrorIsolationEnabled) {
+                throw e;
+            }
+            quarantineUnreadableDirtyResult(path, e);
+            return Optional.empty();
+        }
+    }
+
+    /**
+     * Moves an unreadable dirty result file aside so that it stops being picked up by future
+     * recovery attempts, leaving it for manual inspection/cleanup.
+     */
+    private void quarantineUnreadableDirtyResult(Path path, IOException cause) {
+        final Path quarantinedPath = constructQuarantinedPath(path);
+
+        LOG.error(
+                "Could not read the dirty job result entry {}. Its terminal status and "
+                        + "application cannot be determined, so it cannot be safely recovered. "
+                        + "Moving the entry to {} for manual inspection/cleanup and skipping it "
+                        + "for this and future recovery attempts.",
+                path,
+                quarantinedPath,
+                cause);
+        try {
+            fileSystem.rename(path, quarantinedPath);
+        } catch (IOException renameFailure) {
+            LOG.warn(
+                    "Could not move the unreadable job result entry {} to {}; it will be "
+                            + "retried on the next recovery attempt.",
+                    path,
+                    quarantinedPath,
+                    renameFailure);
+        }
+    }
+
+    private Path constructQuarantinedPath(Path originalPath) {
+        return constructEntryPath(originalPath.getName() + QUARANTINED_FILE_SUFFIX);
+    }
+
     /**
      * Wrapper class around {@link JobResultEntry} to allow for serialization of a schema version,
      * so that future schema changes can be handled in a backwards compatible manner.

flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreFileOperationsTest.java

@@ -38,12 +38,15 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 import static org.apache.flink.runtime.highavailability.JobResultStoreContractTest.DUMMY_JOB_RESULT_ENTRY;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for the internal {@link FileSystemJobResultStore} mechanisms. */
 @ExtendWith(TestLoggerExtension.class)
@@ -268,6 +271,52 @@ public void testJobResultSerializationDeserialization() throws IOException {
                 .isEqualTo(DUMMY_JOB_RESULT_ENTRY.getJobResult().getAccumulatorResults());
     }
 
+    @Test
+    public void testGetDirtyResultsWithCorruptedEntryAndIsolationEnabledQuarantinesFile()
+            throws Exception {
+        fileSystemJobResultStore =
+                new FileSystemJobResultStore(
+                        basePath.getFileSystem(), basePath, false, manuallyTriggeredExecutor, true);
+
+        final JobID jobId = new JobID();
+        final File corruptedDirtyFile = writeCorruptedDirtyFile(jobId);
+
+        // the corrupted entry is skipped entirely -- its terminal status and application can't
+        // be recovered, and this codebase requires every dirty JobResult to have both, so no
+        // placeholder is synthesized for it
+        assertThat(fileSystemJobResultStore.getDirtyResults()).isEmpty();
+
+        assertThat(corruptedDirtyFile).doesNotExist();
+        assertThat(
+                        new File(
+                                temporaryFolder,
+                                corruptedDirtyFile.getName()
+                                        + FileSystemJobResultStore.QUARANTINED_FILE_SUFFIX))
+                .exists();
+
+        // a subsequent recovery attempt no longer trips over the quarantined file
+        assertThat(fileSystemJobResultStore.getDirtyResults()).isEmpty();
+    }
+
+    @Test
+    public void testGetDirtyResultsWithCorruptedEntryAndIsolationDisabledFailsFast()
+            throws Exception {
+        final JobID jobId = new JobID();
+        writeCorruptedDirtyFile(jobId);
+
+        assertThatThrownBy(() -> fileSystemJobResultStore.getDirtyResults())
+                .isInstanceOf(IOException.class);
+    }
+
+    private File writeCorruptedDirtyFile(JobID jobId) throws IOException {
+        final File corruptedDirtyFile =
+                new File(
+                        temporaryFolder,
+                        jobId.toString() + FileSystemJobResultStore.DIRTY_FILE_EXTENSION);
+        Files.write(corruptedDirtyFile.toPath(), "not valid json".getBytes(StandardCharsets.UTF_8));
+        return corruptedDirtyFile;
+    }
+
     private List<JobID> getCleanResultIdsFromFileSystem() throws IOException {
         final List<JobID> cleanResults = new ArrayList<>();