XComp commented on code in PR #22341:
URL: https://github.com/apache/flink/pull/22341#discussion_r1263411729
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStore.java:
##########
@@ -43,69 +43,67 @@ public interface JobResultStore {
* Registers the passed {@link JobResultEntry} instance as {@code dirty}
which indicates that
* clean-up operations still need to be performed. Once the job resource
cleanup has been
* finalized, we can mark the {@code JobResultEntry} as {@code clean}
result using {@link
- * #markResultAsClean(JobID)}.
+ * #markResultAsCleanAsync(JobID)}.
*
* @param jobResultEntry The job result we wish to persist.
- * @throws IOException if the creation of the dirty result failed for IO
reasons.
- * @throws IllegalStateException if the passed {@code jobResultEntry} has
a {@code JobID}
- * attached that is already registered in this {@code JobResultStore}.
+ * @return CompletableFuture it the future with {@code true} if the dirty
result is created
+ * successfully, otherwise will throw {@link IllegalStateException} if
the passed {@code
+ * jobResultEntry} has a {@code JobID} attached that is already
registered in this {@code
+ * JobResultStore}.
*/
- void createDirtyResult(JobResultEntry jobResultEntry) throws IOException,
IllegalStateException;
+ CompletableFuture<Boolean> createDirtyResultAsync(JobResultEntry
jobResultEntry);
Review Comment:
Why did we change from `CompletableFuture<Void>` to
`CompletableFuture<Boolean>`? Boolean doesn't add any value here. Or am I
missing something :thinking:
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStore.java:
##########
@@ -43,69 +43,67 @@ public interface JobResultStore {
* Registers the passed {@link JobResultEntry} instance as {@code dirty}
which indicates that
* clean-up operations still need to be performed. Once the job resource
cleanup has been
* finalized, we can mark the {@code JobResultEntry} as {@code clean}
result using {@link
- * #markResultAsClean(JobID)}.
+ * #markResultAsCleanAsync(JobID)}.
*
* @param jobResultEntry The job result we wish to persist.
- * @throws IOException if the creation of the dirty result failed for IO
reasons.
- * @throws IllegalStateException if the passed {@code jobResultEntry} has
a {@code JobID}
- * attached that is already registered in this {@code JobResultStore}.
+ * @return CompletableFuture it the future with {@code true} if the dirty
result is created
Review Comment:
```suggestion
* @return a successfully completed future with {@code true} if the
dirty result is created
```
nit
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/nonha/embedded/EmbeddedJobResultStore.java:
##########
Review Comment:
`markResultAsCleanInternal` has an obsolete `IOException` declaration in the
method signature
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractThreadsafeJobResultStore.java:
##########
@@ -44,64 +45,87 @@ public abstract class AbstractThreadsafeJobResultStore
implements JobResultStore
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
@Override
- public void createDirtyResult(JobResultEntry jobResultEntry) throws
IOException {
- Preconditions.checkState(
- !hasJobResultEntry(jobResultEntry.getJobId()),
- "Job result store already contains an entry for job %s",
- jobResultEntry.getJobId());
-
- withWriteLock(() -> createDirtyResultInternal(jobResultEntry));
+ public CompletableFuture<Boolean> createDirtyResultAsync(JobResultEntry
jobResultEntry) {
+ return hasJobResultEntryAsync(jobResultEntry.getJobId())
+ .handle(
+ (hasResult, error) -> {
+ if (error != null || hasResult) {
Review Comment:
There's a `ExceptionUtils.tryRethrowException` method that's used to handle
the case of `error` not being null. `hasResult` being false can be kept in the
Precondition that's also used in the original code to reduce the diff. WDYT?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -332,11 +332,16 @@ public void testJobBeingMarkedAsDirtyBeforeCleanup()
throws Exception {
TestingJobResultStore.builder()
.withCreateDirtyResultConsumer(
ignoredJobResultEntry -> {
+ CompletableFuture<Boolean>
result =
+ new
CompletableFuture<>();
try {
markAsDirtyLatch.await();
} catch
(InterruptedException e) {
- throw new
RuntimeException(e);
+
result.completeExceptionally(
+ new
RuntimeException(e));
Review Comment:
We shouldn't swallow the `InteruptedException`. Instead, we should call
`Thread.currentThread().interrupt()` to continue the initiated interruption.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -332,11 +332,16 @@ public void testJobBeingMarkedAsDirtyBeforeCleanup()
throws Exception {
TestingJobResultStore.builder()
.withCreateDirtyResultConsumer(
ignoredJobResultEntry -> {
+ CompletableFuture<Boolean>
result =
+ new
CompletableFuture<>();
try {
markAsDirtyLatch.await();
} catch
(InterruptedException e) {
- throw new
RuntimeException(e);
+
result.completeExceptionally(
+ new
RuntimeException(e));
}
+ result.complete(true);
+ return result;
Review Comment:
```suggestion
return
CompletableFuture.completedFuture(true);
```
Even though, I'm still not sure whether we need a `boolean` return value (in
that case `FutureUtils.completedVoidFuture()` would be a valid factory method
for creating the return value.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -548,7 +557,10 @@ public void
testFatalErrorIfJobCannotBeMarkedDirtyInJobResultStore() throws Exce
TestingJobResultStore.builder()
.withCreateDirtyResultConsumer(
jobResult -> {
- throw new IOException("Expected
IOException.");
+ CompletableFuture<Boolean> future = new
CompletableFuture<>();
+ future.completeExceptionally(
+ new IOException("Expected
IOException."));
+ return future;
Review Comment:
```suggestion
return
FutureUtils.completedExceptionally(new IOException("Expected IOException."))
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -1262,14 +1277,8 @@ private CompletableFuture<Void> removeJob(JobID jobId,
CleanupJobState cleanupJo
}
}
- private void markJobAsClean(JobID jobId) {
- try {
- jobResultStore.markResultAsClean(jobId);
- log.debug(
- "Cleanup for the job '{}' has finished. Job has been
marked as clean.", jobId);
- } catch (IOException e) {
- log.warn("Could not properly mark job {} result as clean.", jobId,
e);
- }
+ private CompletableFuture<Void> markJobAsCleanAsync(JobID jobId) {
Review Comment:
You would rather have a `thenRun` in this method to trigger the debug log
message.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -575,12 +576,13 @@ private boolean isDuplicateJob(JobID jobId) throws
FlinkException {
*/
private boolean isInGloballyTerminalState(JobID jobId) throws
FlinkException {
try {
- return jobResultStore.hasJobResultEntry(jobId);
- } catch (IOException e) {
- throw new FlinkException(
- String.format("Failed to retrieve job scheduling status
for job %s.", jobId),
- e);
+ return jobResultStore.hasJobResultEntryAsync(jobId).get();
Review Comment:
This hasn't been addressed.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/SessionDispatcherLeaderProcess.java:
##########
@@ -191,8 +191,8 @@ private Collection<JobResult> getDirtyJobResultsIfRunning()
{
private Collection<JobResult> getDirtyJobResults() {
try {
- return jobResultStore.getDirtyResults();
- } catch (IOException e) {
+ return jobResultStore.getDirtyResultsAsync().get();
Review Comment:
I added a related comment in `JobDispatcherLeaderProcessFactoryFactory`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStore.java:
##########
@@ -43,69 +43,67 @@ public interface JobResultStore {
* Registers the passed {@link JobResultEntry} instance as {@code dirty}
which indicates that
* clean-up operations still need to be performed. Once the job resource
cleanup has been
* finalized, we can mark the {@code JobResultEntry} as {@code clean}
result using {@link
- * #markResultAsClean(JobID)}.
+ * #markResultAsCleanAsync(JobID)}.
*
* @param jobResultEntry The job result we wish to persist.
- * @throws IOException if the creation of the dirty result failed for IO
reasons.
- * @throws IllegalStateException if the passed {@code jobResultEntry} has
a {@code JobID}
- * attached that is already registered in this {@code JobResultStore}.
+ * @return CompletableFuture it the future with {@code true} if the dirty
result is created
+ * successfully, otherwise will throw {@link IllegalStateException} if
the passed {@code
+ * jobResultEntry} has a {@code JobID} attached that is already
registered in this {@code
+ * JobResultStore}.
*/
- void createDirtyResult(JobResultEntry jobResultEntry) throws IOException,
IllegalStateException;
+ CompletableFuture<Boolean> createDirtyResultAsync(JobResultEntry
jobResultEntry);
/**
* Marks an existing {@link JobResultEntry} as {@code clean}. This
indicates that no more
* resource cleanup steps need to be performed. No actions should be
triggered if the passed
* {@code JobID} belongs to a job that was already marked as clean.
*
* @param jobId Ident of the job we wish to mark as clean.
- * @throws IOException if marking the {@code dirty} {@code JobResultEntry}
as {@code clean}
- * failed for IO reasons.
- * @throws NoSuchElementException if there is no corresponding {@code
dirty} job present in the
+ * @return CompletableFuture is the future with the completed state, which
will throw {@link
Review Comment:
```suggestion
* @return the future with the completed state, which will complete
exceptionally with {@link
```
That's a nitty one as well, but: The future doesn't throw (implying that
it's an actor) but rather fails (being a passive component). The future's
`get()` call would through the exception.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java:
##########
@@ -181,37 +188,70 @@ public void markResultAsCleanInternal(JobID jobId) throws
IOException, NoSuchEle
}
@Override
- public boolean hasDirtyJobResultEntryInternal(JobID jobId) throws
IOException {
- return fileSystem.exists(constructDirtyPath(jobId));
+ public CompletableFuture<Boolean> hasDirtyJobResultEntryInternal(JobID
jobId) {
+ CompletableFuture<Boolean> hasDirtyJobResultEntryFuture = new
CompletableFuture<>();
+ ioExecutor.execute(
+ () -> {
+ try {
+ hasDirtyJobResultEntryFuture.complete(
+ fileSystem.exists(constructDirtyPath(jobId)));
+ } catch (IOException e) {
+ hasDirtyJobResultEntryFuture.completeExceptionally(e);
+ }
+ });
+ return hasDirtyJobResultEntryFuture;
Review Comment:
```suggestion
return FutureUtils.supplyAsync(
() -> fileSystem.exists(constructDirtyPath(jobId)),
ioExecutor);
```
There's a shorter version for these kind of code constructrs.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/JobDispatcherLeaderProcessFactoryFactory.java:
##########
@@ -101,14 +109,9 @@ public static JobDispatcherLeaderProcessFactoryFactory
create(
return new JobDispatcherLeaderProcessFactoryFactory(jobGraphRetriever);
}
- private static Collection<JobResult> getDirtyJobResults(JobResultStore
jobResultStore) {
- try {
- return jobResultStore.getDirtyResults();
- } catch (IOException e) {
- throw new FlinkRuntimeException(
- "Could not retrieve the JobResults of dirty jobs from the
underlying JobResultStore.",
- e);
- }
+ private static CompletableFuture<Set<JobResult>> getDirtyJobResultsAsync(
Review Comment:
This method doesn't really have a usecase anymore. We're just calling
`jobResultStore.getDirtyResultsAsync()` without any extra logic.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -332,11 +332,16 @@ public void testJobBeingMarkedAsDirtyBeforeCleanup()
throws Exception {
TestingJobResultStore.builder()
.withCreateDirtyResultConsumer(
ignoredJobResultEntry -> {
+ CompletableFuture<Boolean>
result =
+ new
CompletableFuture<>();
try {
markAsDirtyLatch.await();
} catch
(InterruptedException e) {
- throw new
RuntimeException(e);
+
result.completeExceptionally(
+ new
RuntimeException(e));
Review Comment:
What's the reason for adding the `RuntimeException` as a wrapper here?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/JobDispatcherLeaderProcessFactoryFactory.java:
##########
@@ -77,7 +78,14 @@ public JobDispatcherLeaderProcessFactory createFactory(
}
final JobResultStore jobResultStore =
jobPersistenceComponentFactory.createJobResultStore();
- final Collection<JobResult> recoveredDirtyJobResults =
getDirtyJobResults(jobResultStore);
+ Collection<JobResult> recoveredDirtyJobResults;
Review Comment:
```suggestion
final Collection<JobResult> recoveredDirtyJobResults;
```
The `final` keyword can stay
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStore.java:
##########
@@ -181,37 +188,70 @@ public void markResultAsCleanInternal(JobID jobId) throws
IOException, NoSuchEle
}
@Override
- public boolean hasDirtyJobResultEntryInternal(JobID jobId) throws
IOException {
- return fileSystem.exists(constructDirtyPath(jobId));
+ public CompletableFuture<Boolean> hasDirtyJobResultEntryInternal(JobID
jobId) {
+ CompletableFuture<Boolean> hasDirtyJobResultEntryFuture = new
CompletableFuture<>();
+ ioExecutor.execute(
+ () -> {
+ try {
+ hasDirtyJobResultEntryFuture.complete(
+ fileSystem.exists(constructDirtyPath(jobId)));
+ } catch (IOException e) {
+ hasDirtyJobResultEntryFuture.completeExceptionally(e);
+ }
+ });
+ return hasDirtyJobResultEntryFuture;
}
@Override
- public boolean hasCleanJobResultEntryInternal(JobID jobId) throws
IOException {
- return fileSystem.exists(constructCleanPath(jobId));
+ public CompletableFuture<Boolean> hasCleanJobResultEntryInternal(JobID
jobId) {
+ CompletableFuture<Boolean> hasCleanJobResultEntryFuture = new
CompletableFuture<>();
Review Comment:
Here you can use `FutureUtils.supplyAsync` as well
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -579,10 +591,17 @@ public void
testErrorHandlingIfJobCannotBeMarkedAsCleanInJobResultStore() throws
final CompletableFuture<JobResultEntry> dirtyJobFuture = new
CompletableFuture<>();
final JobResultStore jobResultStore =
TestingJobResultStore.builder()
-
.withCreateDirtyResultConsumer(dirtyJobFuture::complete)
+ .withCreateDirtyResultConsumer(
+ jobResultEntry -> {
+ dirtyJobFuture.complete(jobResultEntry);
+ return
CompletableFuture.completedFuture(true);
+ })
.withMarkResultAsCleanConsumer(
jobId -> {
- throw new IOException("Expected
IOException.");
+ CompletableFuture<Void> result = new
CompletableFuture<>();
+ result.completeExceptionally(
+ new IOException("Expected
IOException."));
+ return result;
Review Comment:
```suggestion
return
FutureUtils.completedExceptionally(new IOException("Expected IOException."));
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -358,7 +363,11 @@ public void testJobBeingMarkedAsCleanAfterCleanup() throws
Exception {
final JobResultStore jobResultStore =
TestingJobResultStore.builder()
-
.withMarkResultAsCleanConsumer(markAsCleanFuture::complete)
+ .withMarkResultAsCleanConsumer(
+ jobID -> {
+ markAsCleanFuture.complete(jobID);
+ return
CompletableFuture.completedFuture(null);
Review Comment:
nit: `FutureUtils.completedVoidFuture()` doesn't create a new instance. That
doesn't make, admittedly, a difference in this test implementation. I wanted to
mention it, anyway.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/JobResultStore.java:
##########
@@ -43,69 +43,67 @@ public interface JobResultStore {
* Registers the passed {@link JobResultEntry} instance as {@code dirty}
which indicates that
* clean-up operations still need to be performed. Once the job resource
cleanup has been
* finalized, we can mark the {@code JobResultEntry} as {@code clean}
result using {@link
- * #markResultAsClean(JobID)}.
+ * #markResultAsCleanAsync(JobID)}.
*
* @param jobResultEntry The job result we wish to persist.
- * @throws IOException if the creation of the dirty result failed for IO
reasons.
- * @throws IllegalStateException if the passed {@code jobResultEntry} has
a {@code JobID}
- * attached that is already registered in this {@code JobResultStore}.
+ * @return CompletableFuture it the future with {@code true} if the dirty
result is created
+ * successfully, otherwise will throw {@link IllegalStateException} if
the passed {@code
+ * jobResultEntry} has a {@code JobID} attached that is already
registered in this {@code
+ * JobResultStore}.
*/
- void createDirtyResult(JobResultEntry jobResultEntry) throws IOException,
IllegalStateException;
+ CompletableFuture<Boolean> createDirtyResultAsync(JobResultEntry
jobResultEntry);
/**
* Marks an existing {@link JobResultEntry} as {@code clean}. This
indicates that no more
* resource cleanup steps need to be performed. No actions should be
triggered if the passed
* {@code JobID} belongs to a job that was already marked as clean.
*
* @param jobId Ident of the job we wish to mark as clean.
- * @throws IOException if marking the {@code dirty} {@code JobResultEntry}
as {@code clean}
- * failed for IO reasons.
- * @throws NoSuchElementException if there is no corresponding {@code
dirty} job present in the
+ * @return CompletableFuture is the future with the completed state, which
will throw {@link
+ * NoSuchElementException} if there is no corresponding {@code dirty}
job present in the
* store for the given {@code JobID}.
*/
- void markResultAsClean(JobID jobId) throws IOException,
NoSuchElementException;
+ CompletableFuture<Void> markResultAsCleanAsync(JobID jobId);
/**
- * Returns whether the store already contains an entry for a job.
+ * Returns the future of whether the store already contains an entry for a
job.
*
* @param jobId Ident of the job we wish to check the store for.
- * @return {@code true} if a {@code dirty} or {@code clean} {@link
JobResultEntry} exists for
- * the given {@code JobID}; otherwise {@code false}.
- * @throws IOException if determining whether a job entry is present in
the store failed for IO
- * reasons.
+ * @return CompletableFuture with {@code true} if a {@code dirty} or
{@code clean} {@link
+ * JobResultEntry} exists for the given {@code JobID}; otherwise
{@code false}.
*/
- default boolean hasJobResultEntry(JobID jobId) throws IOException {
- return hasDirtyJobResultEntry(jobId) || hasCleanJobResultEntry(jobId);
+ default CompletableFuture<Boolean> hasJobResultEntryAsync(JobID jobId) {
+ return hasDirtyJobResultEntryAsync(jobId)
+ .thenCombine(
+ hasCleanJobResultEntryAsync(jobId),
+ (result1, result2) -> result1 || result2);
}
/**
- * Returns whether the store already contains a {@code dirty} entry for
the given {@code JobID}.
+ * Returns the future of whether the store contains a {@code dirty} entry
for the given {@code
+ * JobID}.
*
* @param jobId Ident of the job we wish to check the store for.
- * @return {@code true}, if a {@code dirty} entry exists for the given
{@code JobID}; otherwise
- * {@code false}.
- * @throws IOException if determining whether a job entry is present in
the store failed for IO
- * reasons.
+ * @return CompletableFuture with value of {@code true}, if a {@code
dirty} entry exists for the
+ * given {@code JobID}; otherwise Completable with value of {@code
false}.
*/
- boolean hasDirtyJobResultEntry(JobID jobId) throws IOException;
+ CompletableFuture<Boolean> hasDirtyJobResultEntryAsync(JobID jobId);
/**
- * Returns whether the store already contains a {@code clean} entry for
the given {@code JobID}.
+ * Returns the future of whether the store contains a {@code clean} entry
for the given {@code
+ * JobID}.
*
* @param jobId Ident of the job we wish to check the store for.
- * @return {@code true}, if a {@code clean} entry exists for the given
{@code JobID}; otherwise
- * {@code false}.
- * @throws IOException if determining whether a job entry is present in
the store failed for IO
- * reasons.
+ * @return CompletableFuture with value of {@code true}, if a {@code
clean} entry exists for the
+ * given {@code JobID}; otherwise Completable with value of {@code
false}.
*/
- boolean hasCleanJobResultEntry(JobID jobId) throws IOException;
+ CompletableFuture<Boolean> hasCleanJobResultEntryAsync(JobID jobId);
/**
- * Get the persisted {@link JobResult} instances that are marked as {@code
dirty}. This is
- * useful for recovery of finalization steps.
+ * Returns the future of persisted {@link JobResult} instances that are
marked as {@code dirty}.
+ * This is useful for recovery of finalization steps.
*
- * @return A set of dirty {@code JobResults} from the store.
- * @throws IOException if collecting the set of dirty results failed for
IO reasons.
+ * @return CompletableFuture with value of a set of dirty {@code
JobResults} from the store.
*/
- Set<JobResult> getDirtyResults() throws IOException;
+ CompletableFuture<Set<JobResult>> getDirtyResultsAsync();
Review Comment:
Just to repeat what I already said in another comment: Looks like we can
revert making this method asynchronous.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java:
##########
@@ -1251,7 +1253,20 @@ private CompletableFuture<Void> removeJob(JobID jobId,
CleanupJobState cleanupJo
if (cleanupJobState.isGlobalCleanup()) {
return globalResourceCleaner
.cleanupAsync(jobId)
- .thenRunAsync(() -> markJobAsClean(jobId), ioExecutor)
+ .thenRunAsync(
Review Comment:
You could use `thenComposeAsync` here to avoid the `.get()` call. I'm
wondering why we're using `Async here instead of just `thenRun` :thinking:
##########
flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreContractTest.java:
##########
@@ -35,6 +35,6 @@ public class FileSystemJobResultStoreContractTest implements
JobResultStoreContr
@Override
public JobResultStore createJobResultStore() throws IOException {
Path path = new Path(temporaryFolder.toURI());
- return new FileSystemJobResultStore(path.getFileSystem(), path, false);
+ return new FileSystemJobResultStore(path.getFileSystem(), path, false,
Runnable::run);
Review Comment:
nit: `Executors.directExecutor()` would be the usual Executor for direct
execution.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/FileSystemJobResultStoreFileOperationsTest.java:
##########
@@ -54,11 +58,19 @@ public class FileSystemJobResultStoreFileOperationsTest {
private Path basePath;
+ private ExecutorService ioExecutor;
+
@BeforeEach
public void setupTest() throws IOException {
basePath = new Path(temporaryFolder.toURI());
+ ioExecutor = Executors.newSingleThreadExecutor();
Review Comment:
I guess, for the test cases in this class, the `directExecutor` is good
enough (rather than creating a new thread). You could also use
`ManuallyTriggeredScheduledExecutor` if you want to verify delay in the
execution.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractThreadsafeJobResultStore.java:
##########
@@ -44,64 +45,84 @@ public abstract class AbstractThreadsafeJobResultStore
implements JobResultStore
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
@Override
- public void createDirtyResult(JobResultEntry jobResultEntry) throws
IOException {
- Preconditions.checkState(
- !hasJobResultEntry(jobResultEntry.getJobId()),
- "Job result store already contains an entry for job %s",
- jobResultEntry.getJobId());
-
- withWriteLock(() -> createDirtyResultInternal(jobResultEntry));
+ public CompletableFuture<Void> createDirtyResultAsync(JobResultEntry
jobResultEntry) {
+ return hasJobResultEntryAsync(jobResultEntry.getJobId())
+ .handle(
+ (hasResult, error) -> {
+ if (error != null || hasResult) {
+ ExceptionUtils.rethrow(error);
+ }
+ try {
+ withWriteLock(() ->
createDirtyResultInternal(jobResultEntry));
Review Comment:
Can't we move the asynchronous handling into
`AbstractThreadsafeJobResultStore`. The `AbstractThreadsafeJobResultStore`
would need the `executor` to be passed as a parameter in its constructor. The
`EmbeddedJobResultStore` could then use `Executors.directExecutor()` which
essentially handles the execution synchronously. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/JobDispatcherLeaderProcessFactoryFactory.java:
##########
@@ -103,8 +103,8 @@ public static JobDispatcherLeaderProcessFactoryFactory
create(
private static Collection<JobResult> getDirtyJobResults(JobResultStore
jobResultStore) {
try {
- return jobResultStore.getDirtyResults();
- } catch (IOException e) {
+ return jobResultStore.getDirtyResultsAsync().get();
Review Comment:
Here, I unfortunately misguided you: You just moved the error handling out
of this method. Instead, the calling code does the error handling
(`getDirtyJobResults` becomes obsoletes as part of this refactoring because it
just calls a single method).
The calling code, indeed, has to wait for the dirty results to be returned
synchronously. I'm wondering whether we should simplify the `JobResultStore`
interface here and revert the async refactoring for this specific method.
Thinking about it, we could even separate the interfaces here: The
`getDirtyJobResult` is used in the `DispatcherLeaderProcessFactoryFactory`
implementations (something like `DirtyJobResultRetriever`). The other (now
asynchronous) methods are used in the `Dispatcher` where we would want to have
asynchronous access (`JobResultStoreAsync`). Anyway, I guess, that's not really
necessary in this PR now. But reverting `getDirtyJobResult` appears to be
reasonable. WDYT?
--
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]
