You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/03/30 09:52:03 UTC
[flink] branch release-1.17 updated: [FLINK-31651][runtime] Improve logging of granting/revoking leadership in JobMasterServiceLeadershipRunner to INFO level
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new b398162c5f0 [FLINK-31651][runtime] Improve logging of granting/revoking leadership in JobMasterServiceLeadershipRunner to INFO level
b398162c5f0 is described below
commit b398162c5f0212505eb3fe6e6d8ff4e2485eb4db
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Wed Mar 29 10:40:54 2023 +0800
[FLINK-31651][runtime] Improve logging of granting/revoking leadership in JobMasterServiceLeadershipRunner to INFO level
This closes #22292
---
.../jobmaster/DefaultJobMasterServiceProcess.java | 5 +++++
.../JobMasterServiceLeadershipRunner.java | 25 ++++++++++++++++------
.../runtime/jobmaster/JobMasterServiceProcess.java | 9 ++++++++
.../jobmaster/TestingJobMasterServiceProcess.java | 17 +++++++++++++++
4 files changed, 50 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcess.java
index 5ee7a30164a..f95a06cca39 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcess.java
@@ -180,6 +180,11 @@ public class DefaultJobMasterServiceProcess
return terminationFuture;
}
+ @Override
+ public UUID getLeaderSessionId() {
+ return leaderSessionId;
+ }
+
@Override
public boolean isInitializedAndRunning() {
synchronized (lock) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
index fd1fedca3b1..9cdca3a1916 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunner.java
@@ -272,7 +272,7 @@ public class JobMasterServiceLeadershipRunner implements JobManagerRunner, Leade
throws FlinkException {
try {
if (jobResultStore.hasJobResultEntry(getJobID())) {
- jobAlreadyDone();
+ jobAlreadyDone(leaderSessionId);
} else {
createNewJobMasterServiceProcess(leaderSessionId);
}
@@ -289,7 +289,12 @@ public class JobMasterServiceLeadershipRunner implements JobManagerRunner, Leade
jobMasterServiceProcessFactory.createArchivedExecutionGraph(jobStatus, null));
}
- private void jobAlreadyDone() {
+ private void jobAlreadyDone(UUID leaderSessionId) {
+ LOG.info(
+ "{} for job {} was granted leadership with leader id {}, but job was already done.",
+ getClass().getSimpleName(),
+ getJobID(),
+ leaderSessionId);
resultFuture.complete(
JobManagerRunnerResult.forSuccess(
new ExecutionGraphInfo(
@@ -302,9 +307,12 @@ public class JobMasterServiceLeadershipRunner implements JobManagerRunner, Leade
private void createNewJobMasterServiceProcess(UUID leaderSessionId) throws FlinkException {
Preconditions.checkState(jobMasterServiceProcess.closeAsync().isDone());
- LOG.debug(
- "Create new JobMasterServiceProcess because we were granted leadership under {}.",
- leaderSessionId);
+ LOG.info(
+ "{} for job {} was granted leadership with leader id {}. Creating new {}.",
+ getClass().getSimpleName(),
+ getJobID(),
+ leaderSessionId,
+ JobMasterServiceProcess.class.getSimpleName());
jobMasterServiceProcess = jobMasterServiceProcessFactory.create(leaderSessionId);
@@ -399,7 +407,12 @@ public class JobMasterServiceLeadershipRunner implements JobManagerRunner, Leade
@GuardedBy("lock")
private CompletableFuture<Void> stopJobMasterServiceProcess() {
- LOG.debug("Stop current JobMasterServiceProcess because the leadership has been revoked.");
+ LOG.info(
+ "{} for job {} was revoked leadership with leader id {}. Stopping current {}.",
+ getClass().getSimpleName(),
+ getJobID(),
+ jobMasterServiceProcess.getLeaderSessionId(),
+ JobMasterServiceProcess.class.getSimpleName());
jobMasterGatewayFuture.completeExceptionally(
new FlinkException(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceProcess.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceProcess.java
index f14a6b58b1c..f2948c7af18 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceProcess.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterServiceProcess.java
@@ -24,6 +24,7 @@ import org.apache.flink.util.concurrent.FutureUtils;
import javax.annotation.Nonnull;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
/** JobMasterServiceProcess is responsible for running a {@link JobMasterService}. */
@@ -33,6 +34,9 @@ public interface JobMasterServiceProcess extends AutoCloseableAsync {
return WaitingForLeadership.INSTANCE;
}
+ /** The leader session id of this process. */
+ UUID getLeaderSessionId();
+
/** True iff the {@link JobMasterService} has been initialized and is running. */
boolean isInitializedAndRunning();
@@ -62,6 +66,11 @@ public interface JobMasterServiceProcess extends AutoCloseableAsync {
return FutureUtils.completedVoidFuture();
}
+ @Override
+ public UUID getLeaderSessionId() {
+ throw new UnsupportedOperationException("Still waiting for the leadership.");
+ }
+
@Override
public boolean isInitializedAndRunning() {
return false;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java
index 82edd4f0ae0..c075d88fa09 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobMasterServiceProcess.java
@@ -20,12 +20,14 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
/** Testing implementation of {@link JobMasterServiceProcess}. */
public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
+ private final UUID leaderSessionId;
private final Supplier<CompletableFuture<Void>> closeAsyncSupplier;
private final Supplier<Boolean> isInitializedAndRunningSupplier;
private final Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier;
@@ -33,11 +35,13 @@ public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
private final Supplier<CompletableFuture<String>> getLeaderAddressFutureSupplier;
private TestingJobMasterServiceProcess(
+ UUID leaderSessionId,
Supplier<CompletableFuture<Void>> closeAsyncSupplier,
Supplier<Boolean> isInitializedAndRunningSupplier,
Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier,
Supplier<CompletableFuture<JobManagerRunnerResult>> getResultFutureSupplier,
Supplier<CompletableFuture<String>> getLeaderAddressFutureSupplier) {
+ this.leaderSessionId = leaderSessionId;
this.closeAsyncSupplier = closeAsyncSupplier;
this.isInitializedAndRunningSupplier = isInitializedAndRunningSupplier;
this.getJobMasterGatewayFutureSupplier = getJobMasterGatewayFutureSupplier;
@@ -50,6 +54,11 @@ public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
return closeAsyncSupplier.get();
}
+ @Override
+ public UUID getLeaderSessionId() {
+ return leaderSessionId;
+ }
+
@Override
public boolean isInitializedAndRunning() {
return isInitializedAndRunningSupplier.get();
@@ -76,6 +85,8 @@ public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
/** Builder for {@link TestingJobMasterServiceProcess}. */
public static final class Builder {
+
+ private UUID leaderSessionId = UUID.randomUUID();
private Supplier<CompletableFuture<Void>> closeAsyncSupplier = unsupportedOperation();
private Supplier<Boolean> isInitializedAndRunningSupplier = unsupportedOperation();
private Supplier<CompletableFuture<JobMasterGateway>> getJobMasterGatewayFutureSupplier =
@@ -122,8 +133,14 @@ public class TestingJobMasterServiceProcess implements JobMasterServiceProcess {
return this;
}
+ public Builder setLeaderSessionId(UUID leaderSessionId) {
+ this.leaderSessionId = leaderSessionId;
+ return this;
+ }
+
public TestingJobMasterServiceProcess build() {
return new TestingJobMasterServiceProcess(
+ leaderSessionId,
closeAsyncSupplier,
isInitializedAndRunningSupplier,
getJobMasterGatewayFutureSupplier,