You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2025/07/09 08:41:22 UTC
(flink) 02/08: [FLINK-37561][events] Report JobStatusChange event
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0a7b8f5841501d7622511e597f23c1cfd62a9007
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri May 16 14:03:58 2025 +0200
[FLINK-37561][events] Report JobStatusChange event
---
.../main/java/org/apache/flink/events/Events.java | 3 +-
.../runtime/executiongraph/JobStatusListener.java | 10 ++++
.../apache/flink/runtime/jobmaster/JobMaster.java | 14 +++++-
.../flink/runtime/jobmaster/JobMasterTest.java | 57 ++++++++++++++++++++--
.../runtime/jobmaster/utils/JobMasterBuilder.java | 12 ++++-
5 files changed, 90 insertions(+), 6 deletions(-)
diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/events/Events.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/events/Events.java
index 3b511165584..11fa80f5d99 100644
--- a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/events/Events.java
+++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/events/Events.java
@@ -20,7 +20,8 @@ package org.apache.flink.events;
/** Class that defines known events in an enum. */
public enum Events {
- CheckpointEvent;
+ CheckpointEvent,
+ JobStatusChangeEvent;
public EventBuilder builder(Class<?> classScope) {
return Event.builder(classScope, name());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/JobStatusListener.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/JobStatusListener.java
index dc30d1870be..5459aed8872 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/JobStatusListener.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/JobStatusListener.java
@@ -32,4 +32,14 @@ public interface JobStatusListener {
* @param timestamp The timestamp when the status transition occurred.
*/
void jobStatusChanges(JobID jobId, JobStatus newJobStatus, long timestamp);
+
+ static JobStatusListener combine(JobStatusListener... jobStatusListeners) {
+ return (jobId, newJobStatus, timestamp) -> {
+ for (JobStatusListener jobStatusListener : jobStatusListeners) {
+ if (jobStatusListener != null) {
+ jobStatusListener.jobStatusChanges(jobId, newJobStatus, timestamp);
+ }
+ }
+ };
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 27bd172398f..604f5e4c9ac 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -27,6 +27,7 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.execution.CheckpointType;
import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.core.failure.FailureEnricher;
+import org.apache.flink.events.Events;
import org.apache.flink.queryablestate.KvStateID;
import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
import org.apache.flink.runtime.blob.BlobWriter;
@@ -380,12 +381,23 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId>
this.failureEnrichers = checkNotNull(failureEnrichers);
+ JobStatusListener schedulerListener =
+ JobStatusListener.combine(
+ jobStatusListener,
+ (jobId, newJobStatus, timestamp) ->
+ jobManagerJobMetricGroup.addEvent(
+ Events.JobStatusChangeEvent.builder(JobMaster.class)
+ .setObservedTsMillis(timestamp)
+ .setSeverity("INFO")
+ .setAttribute(
+ "newJobStatus", newJobStatus.name())));
+
this.schedulerNG =
createScheduler(
slotPoolServiceSchedulerFactory,
executionDeploymentTracker,
jobManagerJobMetricGroup,
- jobStatusListener);
+ schedulerListener);
this.heartbeatServices = checkNotNull(heartbeatServices);
this.taskManagerHeartbeatManager = NoOpHeartbeatManager.getInstance();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 1bd65315b2f..9b2d23dc39c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -36,6 +36,9 @@ import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.core.testutils.OneShotLatch;
+import org.apache.flink.events.Event;
+import org.apache.flink.events.EventBuilder;
+import org.apache.flink.events.Events;
import org.apache.flink.runtime.blocklist.BlockedNode;
import org.apache.flink.runtime.blocklist.DefaultBlocklistHandler;
import org.apache.flink.runtime.checkpoint.CheckpointProperties;
@@ -79,6 +82,7 @@ import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.jobmaster.slotpool.DeclarativeSlotPoolFactory;
import org.apache.flink.runtime.jobmaster.slotpool.FreeSlotTracker;
import org.apache.flink.runtime.jobmaster.slotpool.FreeSlotTrackerTestUtils;
@@ -90,6 +94,8 @@ import org.apache.flink.runtime.jobmaster.slotpool.TestingSlotPoolServiceBuilder
import org.apache.flink.runtime.jobmaster.utils.JobMasterBuilder;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
@@ -121,6 +127,7 @@ import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.streaming.api.graph.ExecutionPlan;
import org.apache.flink.streaming.util.RestartStrategyUtils;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.util.FlinkException;
@@ -1688,16 +1695,31 @@ class JobMasterTest {
*/
@Test
void testJobFailureWhenGracefulTaskExecutorTermination() throws Exception {
+ final List<Event> jobEvents = new ArrayList<>();
runJobFailureWhenTaskExecutorTerminatesTest(
heartbeatServices,
(localTaskManagerLocation, jobMasterGateway) ->
jobMasterGateway.disconnectTaskManager(
localTaskManagerLocation.getResourceID(),
- new FlinkException("Test disconnectTaskManager exception.")));
+ new FlinkException("Test disconnectTaskManager exception.")),
+ jobEvents);
+ assertThat(
+ jobEvents.stream()
+ .filter(
+ event ->
+ Events.JobStatusChangeEvent.name()
+ .equals(event.getName()))
+ .map(Event::getAttributes)
+ .map(x -> x.get("newJobStatus")))
+ .containsExactly(
+ JobStatus.RUNNING.toString(),
+ JobStatus.FAILING.toString(),
+ JobStatus.FAILED.toString());
}
@Test
void testJobFailureWhenTaskExecutorHeartbeatTimeout() throws Exception {
+ final List<Event> jobEvents = new ArrayList<>();
final TestingHeartbeatServices testingHeartbeatService =
new TestingHeartbeatServices(heartbeatInterval, heartbeatTimeout);
@@ -1705,7 +1727,20 @@ class JobMasterTest {
testingHeartbeatService,
(localTaskManagerLocation, jobMasterGateway) ->
testingHeartbeatService.triggerHeartbeatTimeout(
- jmResourceId, localTaskManagerLocation.getResourceID()));
+ jmResourceId, localTaskManagerLocation.getResourceID()),
+ jobEvents);
+ assertThat(
+ jobEvents.stream()
+ .filter(
+ event ->
+ Events.JobStatusChangeEvent.name()
+ .equals(event.getName()))
+ .map(Event::getAttributes)
+ .map(x -> x.get("newJobStatus")))
+ .containsExactly(
+ JobStatus.RUNNING.toString(),
+ JobStatus.FAILING.toString(),
+ JobStatus.FAILED.toString());
}
/**
@@ -2497,7 +2532,8 @@ class JobMasterTest {
private void runJobFailureWhenTaskExecutorTerminatesTest(
HeartbeatServices heartbeatServices,
- BiConsumer<LocalUnresolvedTaskManagerLocation, JobMasterGateway> jobReachedRunningState)
+ BiConsumer<LocalUnresolvedTaskManagerLocation, JobMasterGateway> jobReachedRunningState,
+ List<Event> jobEventsOut)
throws Exception {
final JobGraph jobGraph = JobGraphTestUtils.singleNoOpJobGraph();
final JobMasterBuilder.TestingOnCompletionActions onCompletionActions =
@@ -2509,6 +2545,21 @@ class JobMasterTest {
.withHighAvailabilityServices(haServices)
.withHeartbeatServices(heartbeatServices)
.withOnCompletionActions(onCompletionActions)
+ .withMetricsGroupFactory(
+ new JobManagerJobMetricGroupFactory() {
+ @Override
+ public JobManagerJobMetricGroup create(
+ @Nonnull ExecutionPlan executionPlan) {
+ return new UnregisteredMetricGroups
+ .UnregisteredJobManagerJobMetricGroup() {
+ @Override
+ public void addEvent(EventBuilder eventBuilder) {
+ jobEventsOut.add(
+ eventBuilder.build(getAllVariables()));
+ }
+ };
+ }
+ })
.createJobMaster()) {
jobMaster.start();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/JobMasterBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/JobMasterBuilder.java
index a6d30b41ff8..a23e49990b3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/JobMasterBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/JobMasterBuilder.java
@@ -42,6 +42,7 @@ import org.apache.flink.runtime.jobmaster.JobMasterConfiguration;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.SlotPoolServiceSchedulerFactory;
import org.apache.flink.runtime.jobmaster.TestingJobManagerSharedServicesBuilder;
+import org.apache.flink.runtime.jobmaster.factories.JobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.jobmaster.factories.UnregisteredJobManagerJobMetricGroupFactory;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
@@ -96,6 +97,9 @@ public class JobMasterBuilder {
private BlocklistHandler.Factory blocklistHandlerFactory = new NoOpBlocklistHandler.Factory();
+ private JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory =
+ UnregisteredJobManagerJobMetricGroupFactory.INSTANCE;
+
public JobMasterBuilder(JobGraph jobGraph, RpcService rpcService) {
TestingHighAvailabilityServices testingHighAvailabilityServices =
new TestingHighAvailabilityServices();
@@ -183,6 +187,12 @@ public class JobMasterBuilder {
return this;
}
+ public JobMasterBuilder withMetricsGroupFactory(
+ JobManagerJobMetricGroupFactory jobManagerJobMetricGroupFactory) {
+ this.jobManagerJobMetricGroupFactory = jobManagerJobMetricGroupFactory;
+ return this;
+ }
+
public JobMasterBuilder withJobMasterId(JobMasterId jobMasterId) {
this.jobMasterId = jobMasterId;
return this;
@@ -207,7 +217,7 @@ public class JobMasterBuilder {
? jobManagerSharedServices
: new TestingJobManagerSharedServicesBuilder().build(),
heartbeatServices,
- UnregisteredJobManagerJobMetricGroupFactory.INSTANCE,
+ jobManagerJobMetricGroupFactory,
onCompletionActions,
fatalErrorHandler,
JobMasterBuilder.class.getClassLoader(),