This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e01bbcacd96 [FLINK-34246][runtime] Add option to only archive failed
jobs to history server (#29015)
e01bbcacd96 is described below
commit e01bbcacd962189c726d8324a35d66cc162712e7
Author: Archit Goyal <[email protected]>
AuthorDate: Fri Sep 11 12:10:47 2026 -0700
[FLINK-34246][runtime] Add option to only archive failed jobs to history
server (#29015)
* [FLINK-34246][runtime] Add option to only archive failed jobs to history
server
Adds jobmanager.archive.only-failed-jobs (default false) to skip archiving
finished/canceled/suspended jobs to the history server.
* Add missing positive archiving test and fix config option wording
* Fix new test hanging by marking the application job status as failed
* Set a failure cause on the archived execution graph for failed jobs in
the test to avoid a hang
* Regenerate the jobmanager archive only failed jobs config doc so it
matches the option description
---------
Co-authored-by: argoyal_Linkedin <[email protected]>
---
.../generated/all_jobmanager_section.html | 6 ++
.../generated/job_manager_configuration.html | 6 ++
.../flink/configuration/JobManagerOptions.java | 23 ++++++
.../flink/runtime/dispatcher/Dispatcher.java | 14 +++-
.../dispatcher/DispatcherResourceCleanupTest.java | 92 ++++++++++++++++++++--
5 files changed, 135 insertions(+), 6 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/all_jobmanager_section.html
b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
index 5d01a4a4885..42eb43eea22 100644
--- a/docs/layouts/shortcodes/generated/all_jobmanager_section.html
+++ b/docs/layouts/shortcodes/generated/all_jobmanager_section.html
@@ -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>
diff --git a/docs/layouts/shortcodes/generated/job_manager_configuration.html
b/docs/layouts/shortcodes/generated/job_manager_configuration.html
index 75aa1a1d9c8..1023058a36b 100644
--- a/docs/layouts/shortcodes/generated/job_manager_configuration.html
+++ b/docs/layouts/shortcodes/generated/job_manager_configuration.html
@@ -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>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index a185ce22f67..847678a9e45 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -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}
*/
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index f64645b196b..c29be3274f2 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -33,6 +33,7 @@ import org.apache.flink.configuration.ConfigOptions;
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;
@@ -2315,7 +2316,9 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
// 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));
@@ -2413,6 +2416,15 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
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) {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
index 1d9b3e87475..ee6d146f560 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.ApplicationID;
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;
@@ -32,6 +33,7 @@ import
org.apache.flink.runtime.client.DuplicateJobSubmissionException;
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;
@@ -93,6 +95,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
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}. */
@@ -494,18 +497,31 @@ public class DispatcherResourceCleanupTest extends
TestLogger {
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() {
@@ -736,6 +752,72 @@ public class DispatcherResourceCleanupTest extends
TestLogger {
assertFalse(isArchived.get());
}
+ @Test
+ 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 {
