You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2022/07/04 07:36:41 UTC
[flink] branch master updated: [FLINK-19358][runtime] Make the job id distinct in application mode when HA is enabled.
This is an automated email from the ASF dual-hosted git repository.
guoyangze 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 e70fe68dea7 [FLINK-19358][runtime] Make the job id distinct in application mode when HA is enabled.
e70fe68dea7 is described below
commit e70fe68dea764606180ca3728184c00fc63ea0ff
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Wed Jun 29 10:25:10 2022 +0800
[FLINK-19358][runtime] Make the job id distinct in application mode when HA is enabled.
Previously, the job id will be set to zero in application mode when HA is enabled and the job id is not configured by user.
This commit makes the job id distinct across different jobs in such case, which allows HistoryServer to distinguish it.
This closes #20042.
---
.../ApplicationDispatcherBootstrap.java | 16 ++++++++++---
.../ApplicationDispatcherBootstrapITCase.java | 28 +++++++++-------------
.../ApplicationDispatcherBootstrapTest.java | 15 ++++++++----
flink-end-to-end-tests/test-scripts/common.sh | 2 +-
.../test_ha_per_job_cluster_datastream.sh | 3 ++-
5 files changed, 37 insertions(+), 27 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrap.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrap.java
index 808d6c2fffa..b6ebd8418ca 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrap.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrap.java
@@ -29,6 +29,7 @@ import org.apache.flink.client.deployment.application.executors.EmbeddedExecutor
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.PipelineOptionsInternal;
import org.apache.flink.core.execution.PipelineExecutorServiceLoader;
import org.apache.flink.runtime.client.DuplicateJobSubmissionException;
@@ -42,6 +43,7 @@ import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.concurrent.ScheduledExecutor;
@@ -80,8 +82,6 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
@Internal
public class ApplicationDispatcherBootstrap implements DispatcherBootstrap {
- public static final JobID ZERO_JOB_ID = new JobID(0, 0);
-
@VisibleForTesting static final String FAILED_JOB_NAME = "(application driver)";
private static final Logger LOG = LoggerFactory.getLogger(ApplicationDispatcherBootstrap.class);
@@ -216,8 +216,18 @@ public class ApplicationDispatcherBootstrap implements DispatcherBootstrap {
dispatcherGateway, scheduledExecutor, false, submitFailedJobOnApplicationError);
}
if (!configuredJobId.isPresent()) {
+ // In HA mode, we only support single-execute jobs at the moment. Here, we manually
+ // generate the job id, if not configured, from the cluster id to keep it consistent
+ // across failover.
configuration.set(
- PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, ZERO_JOB_ID.toHexString());
+ PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID,
+ new JobID(
+ Preconditions.checkNotNull(
+ configuration.get(
+ HighAvailabilityOptions.HA_CLUSTER_ID))
+ .hashCode(),
+ 0)
+ .toHexString());
}
return runApplicationAsync(
dispatcherGateway, scheduledExecutor, true, submitFailedJobOnApplicationError);
diff --git a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java
index 3a7cf6b2699..8857fa81084 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java
@@ -102,9 +102,11 @@ public class ApplicationDispatcherBootstrapITCase {
public void testDispatcherRecoversAfterLosingAndRegainingLeadership() throws Exception {
final String blockId = UUID.randomUUID().toString();
final Configuration configuration = new Configuration();
+ final JobID jobId = new JobID();
configuration.set(HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());
configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
configuration.set(ClientOptions.CLIENT_RETRY_PERIOD, Duration.ofMillis(100));
+ configuration.set(PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, jobId.toHexString());
final TestingMiniClusterConfiguration clusterConfiguration =
TestingMiniClusterConfiguration.newBuilder()
.setConfiguration(configuration)
@@ -124,13 +126,12 @@ public class ApplicationDispatcherBootstrapITCase {
cluster.start();
// wait until job is running
- awaitJobStatus(cluster, ApplicationDispatcherBootstrap.ZERO_JOB_ID, JobStatus.RUNNING);
+ awaitJobStatus(cluster, jobId, JobStatus.RUNNING);
// make sure the operator is actually running
BlockingJob.awaitRunning(blockId);
- final CompletableFuture<JobResult> firstJobResult =
- cluster.requestJobResult(ApplicationDispatcherBootstrap.ZERO_JOB_ID);
+ final CompletableFuture<JobResult> firstJobResult = cluster.requestJobResult(jobId);
haServices.revokeDispatcherLeadership();
// make sure the leadership is revoked to avoid race conditions
assertThat(firstJobResult.get())
@@ -139,14 +140,13 @@ public class ApplicationDispatcherBootstrapITCase {
haServices.grantDispatcherLeadership();
// job is suspended, wait until it's running
- awaitJobStatus(cluster, ApplicationDispatcherBootstrap.ZERO_JOB_ID, JobStatus.RUNNING);
+ awaitJobStatus(cluster, jobId, JobStatus.RUNNING);
// unblock processing so the job can finish
BlockingJob.unblock(blockId);
// and wait for it to actually finish
- final JobResult secondJobResult =
- cluster.requestJobResult(ApplicationDispatcherBootstrap.ZERO_JOB_ID).get();
+ final JobResult secondJobResult = cluster.requestJobResult(jobId).get();
assertThat(secondJobResult.isSuccess()).isTrue();
assertThat(secondJobResult.getApplicationStatus())
.isEqualTo(ApplicationStatus.SUCCEEDED);
@@ -161,9 +161,11 @@ public class ApplicationDispatcherBootstrapITCase {
@Test
public void testDirtyJobResultRecoveryInApplicationMode() throws Exception {
final Configuration configuration = new Configuration();
+ final JobID jobId = new JobID();
configuration.set(HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());
configuration.set(DeploymentOptions.TARGET, EmbeddedExecutor.NAME);
configuration.set(ClientOptions.CLIENT_RETRY_PERIOD, Duration.ofMillis(100));
+ configuration.set(PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, jobId.toHexString());
final TestingMiniClusterConfiguration clusterConfiguration =
TestingMiniClusterConfiguration.newBuilder()
.setConfiguration(configuration)
@@ -173,9 +175,7 @@ public class ApplicationDispatcherBootstrapITCase {
// implementation fail to submit the job
final JobResultStore jobResultStore = new EmbeddedJobResultStore();
jobResultStore.createDirtyResult(
- new JobResultEntry(
- TestingJobResultStore.createSuccessfulJobResult(
- ApplicationDispatcherBootstrap.ZERO_JOB_ID)));
+ new JobResultEntry(TestingJobResultStore.createSuccessfulJobResult(jobId)));
final EmbeddedHaServicesWithLeadershipControl haServices =
new EmbeddedHaServicesWithLeadershipControl(EXECUTOR_EXTENSION.getExecutor()) {
@@ -205,14 +205,8 @@ public class ApplicationDispatcherBootstrapITCase {
"The job's main method shouldn't have been succeeded due to a DuplicateJobSubmissionException.")
.hasAtLeastOneElementOfType(DuplicateJobSubmissionException.class);
- assertThat(
- jobResultStore.hasDirtyJobResultEntry(
- ApplicationDispatcherBootstrap.ZERO_JOB_ID))
- .isFalse();
- assertThat(
- jobResultStore.hasCleanJobResultEntry(
- ApplicationDispatcherBootstrap.ZERO_JOB_ID))
- .isTrue();
+ assertThat(jobResultStore.hasDirtyJobResultEntry(jobId)).isFalse();
+ assertThat(jobResultStore.hasCleanJobResultEntry(jobId)).isTrue();
}
@Test
diff --git a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapTest.java b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapTest.java
index 9d6ae78c5b9..f4f9bde130e 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapTest.java
@@ -146,10 +146,12 @@ public class ApplicationDispatcherBootstrapTest {
}
@Test
- public void testJobIdDefaultsToZeroWithHa() throws Throwable {
+ public void testJobIdDefaultsToClusterIdWithHa() throws Throwable {
final Configuration configurationUnderTest = getConfiguration();
+ final String clusterId = "cluster";
configurationUnderTest.set(
HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());
+ configurationUnderTest.set(HighAvailabilityOptions.HA_CLUSTER_ID, clusterId);
final CompletableFuture<JobID> submittedJobId = new CompletableFuture<>();
@@ -166,7 +168,9 @@ public class ApplicationDispatcherBootstrapTest {
applicationFuture.get(TIMEOUT_SECONDS, TimeUnit.SECONDS);
- assertThat(submittedJobId.get(TIMEOUT_SECONDS, TimeUnit.SECONDS), is(new JobID(0L, 0L)));
+ assertThat(
+ submittedJobId.get(TIMEOUT_SECONDS, TimeUnit.SECONDS),
+ is(new JobID(clusterId.hashCode(), 0L)));
}
@Test
@@ -763,13 +767,14 @@ public class ApplicationDispatcherBootstrapTest {
@Test
public void testSubmitFailedJobOnApplicationErrorInHASetup() throws Exception {
final Configuration configuration = getConfiguration();
+ final JobID jobId = new JobID();
configuration.set(HighAvailabilityOptions.HA_MODE, HighAvailabilityMode.ZOOKEEPER.name());
configuration.set(DeploymentOptions.SUBMIT_FAILED_JOB_ON_APPLICATION_ERROR, true);
+ configuration.set(PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, jobId.toHexString());
testSubmitFailedJobOnApplicationError(
configuration,
- (jobId, t) -> {
- Assertions.assertThat(jobId)
- .isEqualTo(ApplicationDispatcherBootstrap.ZERO_JOB_ID);
+ (id, t) -> {
+ Assertions.assertThat(id).isEqualTo(jobId);
Assertions.assertThat(t)
.isInstanceOf(ProgramInvocationException.class)
.hasRootCauseInstanceOf(RuntimeException.class)
diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 41b4e0fccf4..df7cd4c3bf7 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -715,7 +715,7 @@ function wait_num_of_occurence_in_logs {
echo "Waiting for text ${text} to appear ${number} of times in logs..."
while : ; do
- N=$(grep -o "${text}" $FLINK_LOG_DIR/*${logs}*.log* | wc -l)
+ N=$(grep -E -o "${text}" $FLINK_LOG_DIR/*${logs}*.log* | wc -l)
if [ -z $N ]; then
N=0
diff --git a/flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh b/flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
index 90625c9c3bc..b76b71ac891 100755
--- a/flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
+++ b/flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh
@@ -23,7 +23,6 @@ source "$(dirname "$0")"/common_ha.sh
TEST_PROGRAM_JAR_NAME=DataStreamAllroundTestProgram.jar
TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-datastream-allround-test/target/${TEST_PROGRAM_JAR_NAME}
FLINK_LIB_DIR=${FLINK_DIR}/lib
-JOB_ID="00000000000000000000000000000000"
#
# NOTE: This script requires at least Bash version >= 4. Mac OS in 2020 still ships 3.x
@@ -126,6 +125,8 @@ function run_ha_test() {
local neededTaskmanagers=$(( (${PARALLELISM} + ${TASK_SLOTS_PER_TM_HA} - 1) / ${TASK_SLOTS_PER_TM_HA} ))
start_taskmanagers ${neededTaskmanagers}
+ wait_num_of_occurence_in_logs "Job [a-z0-9]+ is submitted." 1 "standalonejob"
+ JOB_ID=$(grep -E -o 'Job [a-z0-9]+ is submitted' $FLINK_LOG_DIR/*standalonejob*.log* | awk '{print $2}')
wait_job_running ${JOB_ID}
# start the watchdog that keeps the number of JMs stable