zentol commented on a change in pull request #19047:
URL: https://github.com/apache/flink/pull/19047#discussion_r824633822
##########
File path:
flink-clients/src/test/java/org/apache/flink/client/deployment/application/ApplicationDispatcherBootstrapITCase.java
##########
@@ -165,6 +168,69 @@ public void
testDispatcherRecoversAfterLosingAndRegainingLeadership() throws Exc
@Test
public void testDirtyJobResultRecoveryInApplicationMode() throws Exception
{
+ // having a dirty entry in the JobResultStore should make the
ApplicationDispatcherBootstrap
+ // implementation fail to submit the job
+ final JobResultStore jobResultStore = new EmbeddedJobResultStore();
+ jobResultStore.createDirtyResult(
+ new JobResultEntry(
+ TestingJobResultStore.createSuccessfulJobResult(
+ ApplicationDispatcherBootstrap.ZERO_JOB_ID)));
+
+ final CompletableFuture<Throwable> fatalError = new
CompletableFuture<>();
+ testJobResultStoreEntryLeadingToDuplicateJobSubmissionException(
+ jobResultStore,
+ ErrorHandlingSubmissionJob.createPackagedProgram(),
+ fatalError::complete);
+
+
FlinkAssertions.assertThatChainOfCauses(ErrorHandlingSubmissionJob.getSubmissionException())
+ .as(
+ "The job's main method shouldn't have been succeeded
due to a DuplicateJobSubmissionException.")
+
.hasAtLeastOneElementOfType(DuplicateJobSubmissionException.class);
+
+ assertThat(fatalError).as("No fatal error should have been
triggered.").isNotCompleted();
+
+ assertThat(
+ jobResultStore.hasDirtyJobResultEntry(
+ ApplicationDispatcherBootstrap.ZERO_JOB_ID))
+ .isFalse();
+ assertThat(
+ jobResultStore.hasCleanJobResultEntry(
+ ApplicationDispatcherBootstrap.ZERO_JOB_ID))
+ .isTrue();
+ }
+
+ @Test
+ public void testWithCleanJobResultInApplicationMode() throws Exception {
+ final JobID jobId = ApplicationDispatcherBootstrap.ZERO_JOB_ID;
+ final JobResultStore jobResultStore = new EmbeddedJobResultStore();
+ jobResultStore.createDirtyResult(
+ new
JobResultEntry(TestingJobResultStore.createSuccessfulJobResult(jobId)));
+ // having a clean entry in the JobResultStore should make the
ApplicationDispatcherBootstrap
+ // implementation fail to submit the job
+ jobResultStore.markResultAsClean(jobId);
+
+ final AtomicReference<Throwable> fatalErrorRef = new
AtomicReference<>();
+ testJobResultStoreEntryLeadingToDuplicateJobSubmissionException(
+ jobResultStore, FailingJob.getProgram(), fatalErrorRef::set);
+
+ final DuplicateJobSubmissionException expectedException =
+
DuplicateJobSubmissionException.ofGloballyTerminatedAndCleaned(jobId);
+ assertThat(fatalErrorRef.get())
+ .isNotNull()
+ .as("A fatal error should have caused the Flink cluster to
shut down.")
Review comment:
if this weren't the case then
testJobResultStoreEntryLeadingToDuplicateJobSubmissionException wouldn't return
in the first place, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
