You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2024/11/05 15:19:01 UTC
(flink) branch master updated: [FLINK-36623] Improve logging in DefaultStateTransitionManager (#25610)
This is an automated email from the ASF dual-hosted git repository.
mapohl 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 21eea0d2dde [FLINK-36623] Improve logging in DefaultStateTransitionManager (#25610)
21eea0d2dde is described below
commit 21eea0d2ddeea14a791c13c9595d8cb54e9f1b0c
Author: Zdenek Tison <12...@users.noreply.github.com>
AuthorDate: Tue Nov 5 16:18:52 2024 +0100
[FLINK-36623] Improve logging in DefaultStateTransitionManager (#25610)
* The JobId is available in AdaptiveScheduler states
* The jobId is added to logs in DefaultStateTransitionManager
* Improved logging in DefaultStateTransitionManager
---
.../scheduler/adaptive/AdaptiveScheduler.java | 5 +++
.../adaptive/DefaultStateTransitionManager.java | 45 +++++++++++++++++-----
.../flink/runtime/scheduler/adaptive/Finished.java | 10 +++--
.../flink/runtime/scheduler/adaptive/State.java | 10 +++++
.../scheduler/adaptive/StateTransitionManager.java | 9 +++++
.../adaptive/StateWithExecutionGraph.java | 3 +-
.../adaptive/StateWithoutExecutionGraph.java | 13 +++++++
.../DefaultStateTransitionManagerTest.java | 7 ++++
.../runtime/scheduler/adaptive/ExecutingTest.java | 5 +++
.../MockStateWithoutExecutionGraphContext.java | 7 ++++
10 files changed, 100 insertions(+), 14 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 040312dba97..6a3f0a4781f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -1133,6 +1133,11 @@ public class AdaptiveScheduler
"Not enough resources available for scheduling."));
}
+ @Override
+ public JobID getJobId() {
+ return jobInfo.getJobId();
+ }
+
@Override
public ArchivedExecutionGraph getArchivedExecutionGraph(
JobStatus jobStatus, @Nullable Throwable cause) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
index 890e8c535ed..5435873be4e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManager.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -109,13 +110,13 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
@Override
public void onChange() {
- LOG.debug("OnChange event received in phase {}.", getPhase());
+ LOG.debug("OnChange event received in phase {} for job {}.", getPhase(), getJobId());
phase.onChange();
}
@Override
public void onTrigger() {
- LOG.debug("OnTrigger event received in phase {}.", getPhase());
+ LOG.debug("OnTrigger event received in phase {} for job {}.", getPhase(), getJobId());
phase.onTrigger();
}
@@ -157,7 +158,7 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
Preconditions.checkState(
!(phase instanceof Transitioning),
"The state transition operation has already been triggered.");
- LOG.info("Transitioning from {} to {}.", phase, newPhase);
+ LOG.info("Transitioning from {} to {}, job {}.", phase, newPhase, getJobId());
phase = newPhase;
}
@@ -172,12 +173,17 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
callback.run();
} else {
LOG.debug(
- "Ignoring scheduled action because expected phase {} is not the actual phase {}.",
+ "Ignoring scheduled action because expected phase {} is not the actual phase {}, job {}.",
expectedPhase,
- getPhase());
+ getPhase(),
+ getJobId());
}
}
+ private JobID getJobId() {
+ return transitionContext.getJobId();
+ }
+
/**
* A phase in the state machine of the {@link DefaultStateTransitionManager}. Each phase is
* responsible for a specific part of the state transition process.
@@ -233,6 +239,10 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
public String toString() {
return getClass().getSimpleName();
}
+
+ JobID getJobId() {
+ return context.getJobId();
+ }
}
/**
@@ -350,11 +360,14 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
private void transitionToSubSequentStateForDesiredResources() {
if (hasDesiredResources()) {
- LOG.info("Desired resources are met, transitioning to the subsequent state.");
+ LOG.info(
+ "Desired resources are met, transitioning to the subsequent state, job {}.",
+ getJobId());
context().triggerTransitionToSubsequentState();
} else {
LOG.debug(
- "Desired resources are not met, skipping the transition to the subsequent state.");
+ "Desired resources are not met, skipping the transition to the subsequent state, job {}.",
+ getJobId());
}
}
}
@@ -373,16 +386,28 @@ public class DefaultStateTransitionManager implements StateTransitionManager {
Temporal firstChangeEventTimestamp,
Duration maxTriggerDelay) {
super(clock, context);
- this.scheduleRelativelyTo(this::onTrigger, firstChangeEventTimestamp, maxTriggerDelay);
+ this.scheduleRelativelyTo(
+ () -> {
+ LOG.info(
+ "Scheduled onTrigger event fired in Stabilized phase, job {}.",
+ getJobId());
+ onTrigger();
+ },
+ firstChangeEventTimestamp,
+ maxTriggerDelay);
}
@Override
void onTrigger() {
if (hasSufficientResources()) {
- LOG.info("Sufficient resources are met, progressing to subsequent state.");
+ LOG.info(
+ "Sufficient resources are met, progressing to subsequent state, job {}.",
+ getJobId());
context().triggerTransitionToSubsequentState();
} else {
- LOG.debug("Sufficient resources are not met, progressing to idling.");
+ LOG.debug(
+ "Sufficient resources are not met, progressing to idling, job {}.",
+ getJobId());
context().progressToIdling();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
index f47dce0441a..87f4d1f0288 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
@@ -56,13 +57,16 @@ class Finished implements State {
return archivedExecutionGraph;
}
+ @Override
+ public JobID getJobId() {
+ return archivedExecutionGraph.getJobID();
+ }
+
@Override
public void handleGlobalFailure(
Throwable cause, CompletableFuture<Map<String, String>> failureLabels) {
logger.debug(
- "Ignore global failure because we already finished the job {}.",
- archivedExecutionGraph.getJobID(),
- cause);
+ "Ignore global failure because we already finished the job {}.", getJobId(), cause);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
index 7d7de39a439..1185274fbd7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.util.function.FunctionWithException;
@@ -51,6 +52,15 @@ interface State extends LabeledGlobalFailureHandler {
*/
void suspend(Throwable cause);
+ /**
+ * Gets the {@link JobID} of the job. The implementation should avoid to use the {@link
+ * State#getJob()} method as it may create the {@link ArchivedExecutionGraph} which is
+ * expensive.
+ *
+ * @return the {@link JobID} of the job
+ */
+ JobID getJobId();
+
/**
* Gets the current {@link JobStatus}. The returned job status will remain unchanged at least
* until the scheduler transitions to a different state.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
index 322e8aae373..5b6f78ab82c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitionManager.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
+
import java.time.Duration;
import java.util.concurrent.ScheduledFuture;
@@ -68,5 +70,12 @@ public interface StateTransitionManager {
* @return a ScheduledFuture representing pending completion of the operation.
*/
ScheduledFuture<?> scheduleOperation(Runnable callback, Duration delay);
+
+ /**
+ * Gets the {@link JobID} of the job.
+ *
+ * @return the {@link JobID} of the job
+ */
+ JobID getJobId();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
index f16487fe3c1..8be4a99f0fd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
@@ -141,7 +141,8 @@ abstract class StateWithExecutionGraph implements State {
return executionGraph;
}
- JobID getJobId() {
+ @Override
+ public JobID getJobId() {
return executionGraph.getJobID();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithoutExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithoutExecutionGraph.java
index 16769e310df..43425445a4c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithoutExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithoutExecutionGraph.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
@@ -53,6 +54,11 @@ abstract class StateWithoutExecutionGraph implements State {
context.goToFinished(context.getArchivedExecutionGraph(JobStatus.SUSPENDED, cause));
}
+ @Override
+ public JobID getJobId() {
+ return context.getJobId();
+ }
+
@Override
public ArchivedExecutionGraph getJob() {
return context.getArchivedExecutionGraph(getJobStatus(), null);
@@ -72,6 +78,13 @@ abstract class StateWithoutExecutionGraph implements State {
/** Context of the {@link StateWithoutExecutionGraph} state. */
interface Context extends StateTransitions.ToFinished {
+ /**
+ * Gets the {@link JobID} of the job.
+ *
+ * @return the {@link JobID} of the job
+ */
+ JobID getJobId();
+
/**
* Creates the {@link ArchivedExecutionGraph} for the given job status and cause. Cause can
* be null if there is no failure.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
index b4ea6720742..885850f9b74 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/DefaultStateTransitionManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.core.testutils.ScheduledTask;
import org.apache.flink.runtime.scheduler.adaptive.DefaultStateTransitionManager.Idling;
import org.apache.flink.util.Preconditions;
@@ -464,6 +465,7 @@ class DefaultStateTransitionManagerTest {
// Instant.MIN makes debugging easier because timestamps become human-readable
private final Instant initializationTime = Instant.MIN;
private Duration elapsedTime = Duration.ZERO;
+ private final JobID jobId = new JobID();
// ///////////////////////////////////////////////
// Context creation
@@ -537,6 +539,11 @@ class DefaultStateTransitionManagerTest {
return scheduledTask;
}
+ @Override
+ public JobID getJobId() {
+ return jobId;
+ }
+
// ///////////////////////////////////////////////
// Test instance creation
// ///////////////////////////////////////////////
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
index 2d00e237bb9..7fb25cc3818 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/ExecutingTest.java
@@ -1038,6 +1038,11 @@ class ExecutingTest {
@Override
public void suspend(Throwable cause) {}
+ @Override
+ public JobID getJobId() {
+ return null;
+ }
+
@Override
public JobStatus getJobStatus() {
return null;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockStateWithoutExecutionGraphContext.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockStateWithoutExecutionGraphContext.java
index a397f07ad5a..07706dae30b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockStateWithoutExecutionGraphContext.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockStateWithoutExecutionGraphContext.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
@@ -38,6 +39,7 @@ class MockStateWithoutExecutionGraphContext
new StateValidator<>("Finished");
private boolean hasStateTransition = false;
+ private final JobID jobId = new JobID();
public void setExpectFinished(Consumer<ArchivedExecutionGraph> asserter) {
finishedStateValidator.expectInput(asserter);
@@ -49,6 +51,11 @@ class MockStateWithoutExecutionGraphContext
registerStateTransition();
}
+ @Override
+ public JobID getJobId() {
+ return jobId;
+ }
+
@Override
public ArchivedExecutionGraph getArchivedExecutionGraph(
JobStatus jobStatus, @Nullable Throwable cause) {