From 6be16c69247393632609859193adcbd3f816fcc3 Mon Sep 17 00:00:00 2001 From: Matthias Pohl Date: Fri, 4 Sep 2026 15:12:22 +0200 Subject: [PATCH] [FLINK-40555][runtime] Isolate unreadable JobResultStore entries per job A corrupted JobResultStore file would have lead to a JobManager failover. The new implementation utilizes the new configuration parameter introduced with FLINK-40554 to skips parsing the entry and quarantines the file under a different name for manual investigation and cleanup. --- .../FileSystemJobResultStore.java | 84 +++++++++++++++++-- ...ystemJobResultStoreFileOperationsTest.java | 49 +++++++++++ 2 files changed, 128 insertions(+), 5 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java index fe98bfd75824ce..a87a626f553df8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java +++ b/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 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 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. diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreFileOperationsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreFileOperationsTest.java index 4e50c1d414f2f4..bfe1d454b6a2a1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreFileOperationsTest.java +++ b/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 getCleanResultIdsFromFileSystem() throws IOException { final List cleanResults = new ArrayList<>();