Skip to content

[FLINK-34246][runtime] Add option to only archive failed jobs to history server #29015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rmetzger merged 5 commits into from
Sep 11, 2026
Merged

[FLINK-34246][runtime] Add option to only archive failed jobs to history server #29015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f6c4f26
[FLINK-34246][runtime] Add option to only archive failed jobs to hist…
argoyal2212 Aug 25, 2026
42336d6
Add missing positive archiving test and fix config option wording
architgyl Sep 10, 2026
2edd490
Fix new test hanging by marking the application job status as failed
architgyl Sep 10, 2026
74bb8ac
Set a failure cause on the archived execution graph for failed jobs i…
architgyl Sep 11, 2026
b69e609
Regenerate the jobmanager archive only failed jobs config doc so it m…
architgyl Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/all_jobmanager_section.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,12 @@
<td>String</td>
<td>Directory for JobManager to store the archives of completed jobs.</td>
</tr>
<tr>
<td><h5>jobmanager.archive.only-failed-jobs</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>Whether to only archive jobs that reached the <code class="highlighter-rouge">FAILED</code> terminal state to <code class="highlighter-rouge">jobmanager.archive.fs.dir</code>. When enabled, jobs that finished or were canceled are not archived to the history server, reducing the number of files written for large clusters running many short-lived batch jobs. This option has no effect unless <code class="highlighter-rouge">jobmanager.archive.fs.dir</code> is configured.</td>
</tr>
<tr>
<td><h5>jobmanager.bind-host</h5></td>
<td style="word-wrap: break-word;">(none)</td>
Expand Down
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/job_manager_configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,12 @@
<td>String</td>
<td>Directory for JobManager to store the archives of completed jobs.</td>
</tr>
<tr>
<td><h5>jobmanager.archive.only-failed-jobs</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>Whether to only archive jobs that reached the <code class="highlighter-rouge">FAILED</code> terminal state to <code class="highlighter-rouge">jobmanager.archive.fs.dir</code>. When enabled, jobs that finished or were canceled are not archived to the history server, reducing the number of files written for large clusters running many short-lived batch jobs. This option has no effect unless <code class="highlighter-rouge">jobmanager.archive.fs.dir</code> is configured.</td>
</tr>
<tr>
<td><h5>jobmanager.bind-host</h5></td>
<td style="word-wrap: break-word;">(none)</td>
Expand Down
23 changes: 23 additions & 0 deletions flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -308,6 +308,29 @@ public class JobManagerOptions {
.withDescription(
"Directory for JobManager to store the archives of completed jobs.");

/**
* Whether only jobs that reached the {@code FAILED} terminal state should be archived to {@link
* #ARCHIVE_DIR}.
*/
@Documentation.Section(Documentation.Sections.ALL_JOB_MANAGER)
public static final ConfigOption<Boolean> ARCHIVE_ON_FAILED_JOBS_ONLY =
key("jobmanager.archive.only-failed-jobs")
.booleanType()
.defaultValue(false)
.withDescription(
Description.builder()
.text(
"Whether to only archive jobs that reached the %s terminal state to %s. ",
code("FAILED"), code(ARCHIVE_DIR.key()))
.text(
"When enabled, jobs that finished or were canceled are not "
+ "archived to the history server, reducing the number of files written "
+ "for large clusters running many short-lived batch jobs. ")
.text(
"This option has no effect unless %s is configured.",
code(ARCHIVE_DIR.key()))
.build());

/**
* @deprecated Use {@link JobManagerOptions#COMPLETED_APPLICATION_STORE_CACHE_SIZE}
*/
Expand Down
14 changes: 13 additions & 1 deletion flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.ThreadDumpMode;
import org.apache.flink.configuration.WebOptions;
Expand Down Expand Up @@ -2315,7 +2316,9 @@ protected CompletableFuture<CleanupJobState> jobReachedTerminalState(
// do not create an archive for suspended jobs, as this would eventually lead to
// multiple archive attempts which we currently do not support
CompletableFuture<Acknowledge> archiveFuture =
archiveExecutionGraphToHistoryServer(executionGraphInfo);
shouldArchiveToHistoryServer(terminalJobStatus)
? archiveExecutionGraphToHistoryServer(executionGraphInfo)
: CompletableFuture.completedFuture(Acknowledge.get());

return archiveFuture.thenCompose(
ignored -> registerGloballyTerminatedJobInJobResultStore(executionGraphInfo));
Expand Down Expand Up @@ -2413,6 +2416,15 @@ private void writeToExecutionGraphInfoStore(ExecutionGraphInfo executionGraphInf
partialExecutionGraphInfoStore.put(executionGraphInfo.getJobId(), executionGraphInfo);
}

/**
* Checks whether a job that reached the given globally terminal state should be archived to the
* history server, honoring {@link JobManagerOptions#ARCHIVE_ON_FAILED_JOBS_ONLY}.
*/
private boolean shouldArchiveToHistoryServer(JobStatus terminalJobStatus) {
return terminalJobStatus == JobStatus.FAILED
|| !configuration.get(JobManagerOptions.ARCHIVE_ON_FAILED_JOBS_ONLY);
}

private CompletableFuture<Acknowledge> archiveExecutionGraphToHistoryServer(
ExecutionGraphInfo executionGraphInfo) {

Expand Down
92 changes: 87 additions & 5 deletions ...time/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.failure.FailureEnricher;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.application.SingleJobApplication;
Expand All @@ -32,6 +33,7 @@
import org.apache.flink.runtime.client.JobSubmissionException;
import org.apache.flink.runtime.dispatcher.cleanup.TestingResourceCleanerFactory;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
import org.apache.flink.runtime.executiongraph.JobStatusListener;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
Expand Down Expand Up @@ -93,6 +95,7 @@
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/** Tests the resource cleanup by the {@link Dispatcher}. */
Expand Down Expand Up @@ -494,18 +497,31 @@ private void suspendJob(TestingJobManagerRunner takeCreatedJobManagerRunner) {
terminateJobWithState(takeCreatedJobManagerRunner, JobStatus.SUSPENDED);
}

private void failJobAndApplication(TestingJobManagerRunner takeCreatedJobManagerRunner) {
terminateJobWithState(takeCreatedJobManagerRunner, JobStatus.FAILED);
application.jobStatusChanges(
takeCreatedJobManagerRunner.getJobID(),
JobStatus.FAILED,
System.currentTimeMillis());
}

private void cancelJob(TestingJobManagerRunner takeCreatedJobManagerRunner) {
terminateJobWithState(takeCreatedJobManagerRunner, JobStatus.CANCELED);
}

private void terminateJobWithState(
TestingJobManagerRunner takeCreatedJobManagerRunner, JobStatus state) {
final ArchivedExecutionGraphBuilder archivedExecutionGraphBuilder =
new ArchivedExecutionGraphBuilder().setJobID(jobId).setState(state);

if (state == JobStatus.FAILED) {
archivedExecutionGraphBuilder.setFailureCause(
new ErrorInfo(
new FlinkException("Test job failure"), System.currentTimeMillis()));
}

takeCreatedJobManagerRunner.completeResultFuture(
new ExecutionGraphInfo(
new ArchivedExecutionGraphBuilder()
.setJobID(jobId)
.setState(state)
.build()));
new ExecutionGraphInfo(archivedExecutionGraphBuilder.build()));
}

private void assertThatNoCleanupWasTriggered() {
Expand Down Expand Up @@ -736,6 +752,72 @@ public void testNotArchivingSuspendedJobToHistoryServer() throws Exception {
assertFalse(isArchived.get());
}

@Test
Comment thread
argoyal2212 marked this conversation as resolved.
public void testNotArchivingFinishedJobToHistoryServerWhenOnlyFailedJobsConfigured()
throws Exception {

final AtomicBoolean isArchived = new AtomicBoolean(false);

final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.ARCHIVE_ON_FAILED_JOBS_ONLY, true);

final TestingDispatcher.Builder testingDispatcherBuilder =
createTestingDispatcherBuilder()
.setConfiguration(configuration)
.setHistoryServerArchivist(
TestingHistoryServerArchivist.builder()
.setArchiveExecutionGraphFunction(
(executionGraphInfo, applicationId) -> {
isArchived.set(true);
return CompletableFuture.completedFuture(
Acknowledge.get());
})
.build());

final TestingJobManagerRunnerFactory jobManagerRunnerFactory =
startDispatcherAndSubmitApplication(testingDispatcherBuilder, 0);

finishJobAndApplication(jobManagerRunnerFactory.takeCreatedJobManagerRunner());

assertGlobalCleanupTriggered(jobId);
dispatcher.getJobTerminationFuture(jobId, Duration.ofHours(1)).join();

assertFalse(isArchived.get());
}

@Test
public void testArchivingFailedJobToHistoryServerWhenOnlyFailedJobsConfigured()
throws Exception {

final AtomicBoolean isArchived = new AtomicBoolean(false);

final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.ARCHIVE_ON_FAILED_JOBS_ONLY, true);

final TestingDispatcher.Builder testingDispatcherBuilder =
createTestingDispatcherBuilder()
.setConfiguration(configuration)
.setHistoryServerArchivist(
TestingHistoryServerArchivist.builder()
.setArchiveExecutionGraphFunction(
(executionGraphInfo, applicationId) -> {
isArchived.set(true);
return CompletableFuture.completedFuture(
Acknowledge.get());
})
.build());

final TestingJobManagerRunnerFactory jobManagerRunnerFactory =
startDispatcherAndSubmitApplication(testingDispatcherBuilder, 0);

failJobAndApplication(jobManagerRunnerFactory.takeCreatedJobManagerRunner());

assertGlobalCleanupTriggered(jobId);
dispatcher.getJobTerminationFuture(jobId, Duration.ofHours(1)).join();

assertTrue(isArchived.get());
}

private static final class BlockingJobManagerRunnerFactory
extends TestingJobMasterServiceLeadershipRunnerFactory {

Expand Down