You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org> on 2024/07/05 16:25:13 UTC
[PR] [FLINK-27355][dispatcher] Unregister JobManagerRunner after it's closed [flink]
kumar-mallikarjuna opened a new pull request, #25027:
URL: https://github.com/apache/flink/pull/25027
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Prevents early unregistration of `JobManagerRunner` before it's closed successfully.
## Brief change log
Moves the `JobManagerRunner` unregistration to another stage in the `CompletableFuture` in `DefaultJobManagerRunnerRegistry.localCleanupAsync()` to keep `JobManagerRunner` from being unregistered before it's closed successfully.
## Verifying this change
The existing tests have been modified to check the new behaviour.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1673394381
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
I see! Is the `unusedExecutor` here always an instance of the class `MainThreadExector`? If so, we can call `thenAccept` and then use this executor to `unregister` on the `main-thread`. 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1680520986
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Adding setters and maintaining nullable state here hints towards a not well-designed class structure.
We're trying to solve the issue that the `DefaultJobManagerRunnerRegistry` requires the main thread executor but still want the main thread verification being executed in a different class (`OnMainThreadJobManagerRunnerRegistry`). That doesn't sound right. We might want to merge the two. But this comes with some problems when initializing the `Dispatcher`.
Can you take a step back and try to reorganize the `Dispatcher` instantiation logic? One hint here: We created the [base Dispatcher constructor](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L259) to improve testability, i.e. to use a testing implementation of `ResourceCleanerFactory`. So, we might be required to rethink about how we initialize the `ResourceCleanerFactory` if cannot keep the `DefaultJobManagerRunnerRegistry` and `OnMainThreadJobManagerRunnerRegistry` separate.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
The main problem here seems to be the constructor chain we have in `Dispatcher`.
Here are the two non-private constructors:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
I see three approaches to do this:
1. [Initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor in say an `initDispatcher(...)` method - probably not a good idea since the attributes would need to be not `final`.
Instead of having the constructor chain, we would have two "base" constructors for `Dispatcher` only:
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
2. We can have two separate constructors which have redundant initializations of the attributes.
3. Constructor (I) would call (II) as:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices) {
this(
rpcService,
fencingToken,
recoveredJobs,
recoveredDirtyJobs,
dispatcherBootstrapFactory,
dispatcherServices,
null,
null);
}
```
In the base constructor, we'll initialize `jobManagerRunnerRegistry/resourceCleanerFactory` when they're `null`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1685926520
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
> But what we can do is making the main thread passing an implementation detail of the `DispatcherResourceCleanerFactory`: We could introduce a twin interface `LocallyCleanableWithMainThreadResource`:
>
> ```java
> @FunctionalInterface
> public interface LocallyCleanableInMainThreadResource {
>
> CompletableFuture<Void> localCleanupAsync(
> JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
> }
> ```
>
> That interface can be converted to a `LocallyCleanableResource` in `DispatcherResourceCleanerFactory`. We wouldn't have to change the `Dispatcher` instantiation in that case. WDYT?
I didn't get this. Which classes would implement `LocallyCleanableInMainThreadResource`? Can you elaborate more?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
The main problem here seems to be the constructor chain we have in `Dispatcher`.
Here are the two non-private constructors:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
I see three approaches to do this:
1. [Initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor in say an `initDispatcher(...)` method - probably not a good idea since the attributes would need to be not `final`.
Instead of having the constructor chain, we would have two "base" constructors for `Dispatcher` only:
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
2. We can have two separate constructors which have redundant initializations of the attributes.
3. Constructor (I) would call (II) as:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices) {
this(
rpcService,
fencingToken,
recoveredJobs,
recoveredDirtyJobs,
dispatcherBootstrapFactory,
dispatcherServices,
null,
null);
}
```
In the base constructor, we'll initialize `jobManagerRunnerRegistry/resourceCleanerFactory` when they're null.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][dispatcher] Unregister JobManagerRunner after it's closed [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2211143608
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c2ac08c0591baabacb15ec24e04456b7bc730fdc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c2ac08c0591baabacb15ec24e04456b7bc730fdc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c2ac08c0591baabacb15ec24e04456b7bc730fdc UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771279162
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/OnMainThreadJobManagerRunnerRegistry.java:
##########
@@ -84,9 +84,10 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor executor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, Executor executor) {
Review Comment:
> I'm wondering whether we should even use ComponentMainThreadExecutor as a type for the main thread executor to make things clear.
I tried this but not sure why but the tests got stuck.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
The main problem here seems to be the constructor chain we have in `Dispatcher`.
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea (since the attributes must be `final`)?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
Alternatively, we can have two separate constructors which have redundant initializations of the attributes, if they must be `final`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1683288457
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
You mean combing Factory and Builder patterns for `DispatcherResourceCleanerFactory`? How about passing the `JobManagerRunnerRegistry` along with `ComponentMainThreadExecutor` in the creator methods there?
```java
public ResourceCleaner createLocalResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor,
+ JobManagerRunnerRegistry jobManagerRunnerRegistry);
```
```java
public ResourceCleaner createGlobalResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor,
+ JobManagerRunnerRegistry jobManagerRunnerRegistry);
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2261134869
> The test `testJobSubmissionUnderSameJobId` fails with this change because we it expects the unregistration to be completed before the `JobManagerRunner` is terminated. Need to look into this.
The test is fixed now.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1687470490
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,15 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApply(
Review Comment:
```suggestion
return resultFuture.thenApplyAsync(
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -30,16 +30,17 @@
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
- private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private final Collection<LocallyCleanableInMainThreadResource>
+ locallyCleanableInMainThreadResource;
private final Collection<GloballyCleanableResource> globallyCleanableResources;
private final Executor cleanupExecutor;
private TestingResourceCleanerFactory(
- Collection<LocallyCleanableResource> locallyCleanableResources,
+ Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResource,
Review Comment:
`LocallyCleanableInMainThreadResource` is an implementation detail of the `DispatcherResourceCleanerFactory`. We don't have to use it in `TestingResourceCleanerFactory`.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
Review Comment:
I'm wondering whether you could come up with a test where we verify that the close is called in another thread and the unregister is then executed in the main thread.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
+ */
+@FunctionalInterface
+public interface LocallyCleanableInMainThreadResource {
+
+ /**
+ * {@code localCleanupAsync} is expected to be called from the main thread and uses the passed
+ * {@code mainThreadExecutor} for cleanups. Thread-safety must be ensured.
+ *
+ * @param jobId The {@link JobID} of the job for which the local data should be cleaned up.
+ * @param mainThreadExecutor The main thread executor.
+ * @return The cleanup result future.
+ */
+ CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor mainThreadExecutor);
Review Comment:
We shouldn't remove the `ioExecutor` from the signature because we still need to translate `LocallyCleanableInMainThreadResource` into `LocallyCleanableResource` (see the proposal in [my comment](https://github.com/apache/flink/pull/25027#discussion_r1683903323)).
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/DispatcherResourceCleanerFactory.java:
##########
@@ -126,8 +129,24 @@ JOB_MANAGER_METRIC_GROUP_LABEL, ofLocalResource(jobManagerMetricGroup))
* @param localResource Local resource that we want to clean during a global cleanup.
* @return Globally cleanable resource.
*/
- private static GloballyCleanableResource ofLocalResource(
+ private static GloballyCleanableResource toGloballyCleanableResource(
+ LocallyCleanableInMainThreadResource localResource) {
+ return localResource::localCleanupAsync;
+ }
+
+ private static GloballyCleanableResource toGloballyCleanableResource(
LocallyCleanableResource localResource) {
return localResource::localCleanupAsync;
}
+
+ /**
+ * Converts a LocallyCleanableInMainThreadResource object to a LocallyCleanableResource object.
+ *
+ * @param localResource LocallyCleanableInMainThreadResource that we want to translate.
+ * @return A LocallyCleanableResource.
+ */
+ private static LocallyCleanableResource toLocallyCleanableResource(
+ LocallyCleanableInMainThreadResource localResource) {
+ return localResource::localCleanupAsync;
Review Comment:
These conversion methods don't have any effect, meaning they wouldn't pass in the main thread in any way. You're just replacing one method with another method having the same signature (but different method and parameter names). The input stays the same (meaning that the `ioExecutor` would be passed in)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771262937
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,42 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources;
private final Collection<GloballyCleanableResource> globallyCleanableResources;
private final Executor cleanupExecutor;
private TestingResourceCleanerFactory(
Collection<LocallyCleanableResource> locallyCleanableResources,
+ Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources,
Collection<GloballyCleanableResource> globallyCleanableResources,
Executor cleanupExecutor) {
this.locallyCleanableResources = locallyCleanableResources;
+ this.locallyCleanableInMainThreadResources = locallyCleanableInMainThreadResources;
this.globallyCleanableResources = globallyCleanableResources;
this.cleanupExecutor = cleanupExecutor;
}
@Override
public ResourceCleaner createLocalResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor) {
+ Collection<LocallyCleanableResource> locallyCleanableResources =
+ this.locallyCleanableResources;
+
+ locallyCleanableResources.addAll(
+ locallyCleanableInMainThreadResources.stream()
+ .map(
+ r ->
+ DispatcherResourceCleanerFactory.toLocallyCleanableResource(
Review Comment:
I didn't get this. By internally you mean have a `toLocallyCleanableResource` method in `TestingResourceCleanerFactory`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771349785
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
+ */
+@FunctionalInterface
+public interface LocallyCleanableInMainThreadResource {
+
+ /**
+ * {@code localCleanupAsync} is expected to be called from the main thread and uses the passed
+ * {@code mainThreadExecutor} for cleanups. Thread-safety must be ensured.
+ *
+ * @param jobId The {@link JobID} of the job for which the local data should be cleaned up.
+ * @param cleanupExecutor The fallback executor for IO-heavy operations.
+ * @param mainThreadExecutor The main thread executor.
+ * @return The cleanup result future.
+ */
+ CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
Review Comment:
yes 👍
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1668650443
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
you're correct - that's what happens in the general scenario (there's an edge case where the CompletableFuture is already completed when chaining the callback to the future instance).
Executing the `unregister(JobID)` in the thread that executed `closeAsync` is something we don't want to do (for the reasons around main thread execution which I outlined in my comment above), though.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2857255146
This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1741038687
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java:
##########
@@ -369,7 +377,9 @@ private MiniDispatcher createMiniDispatcher(
// JobManagerRunnerRegistry needs to be added explicitly
// because cleaning it will trigger the closeAsync latch
// provided by TestingJobManagerRunner
- .withLocallyCleanableResource(jobManagerRunnerRegistry)
+ .withLocallyCleanableResource(
Review Comment:
Correct, thanks!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two param in constructor (I).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2260921569
The test `testJobSubmissionUnderSameJobId` fails with this change because we it expects the unregistration to be completed before the `JobManagerRunner` is terminated. Need to look into this.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1692695511
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,15 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApply(
Review Comment:
I'm curious about this. Do we need to do `thenApplyAsync`, since the function just schedules a task - and isn't blocking? 🤔
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1679615203
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -41,10 +42,12 @@
public class DefaultJobManagerRunnerRegistry implements JobManagerRunnerRegistry {
@VisibleForTesting final Map<JobID, JobManagerRunner> jobManagerRunners;
+ private ComponentMainThreadExecutor mainThreadExecutor;
public DefaultJobManagerRunnerRegistry(int initialCapacity) {
Preconditions.checkArgument(initialCapacity > 0);
jobManagerRunners = CollectionUtil.newHashMapWithExpectedSize(initialCapacity);
+ mainThreadExecutor = null;
Review Comment:
Would need a `null` check or we initialize it to `Runnable::run`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #25027: [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed
URL: https://github.com/apache/flink/pull/25027
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771276366
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
+ */
+@FunctionalInterface
+public interface LocallyCleanableInMainThreadResource {
+
+ /**
+ * {@code localCleanupAsync} is expected to be called from the main thread and uses the passed
+ * {@code mainThreadExecutor} for cleanups. Thread-safety must be ensured.
+ *
+ * @param jobId The {@link JobID} of the job for which the local data should be cleaned up.
+ * @param cleanupExecutor The fallback executor for IO-heavy operations.
+ * @param mainThreadExecutor The main thread executor.
+ * @return The cleanup result future.
+ */
+ CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
Review Comment:
You mean a default method in `LocallyCleanableInMainThreadResource`?
```java
default LocallyCleanableResource toLocallyCleanableResource(
LocallyCleanableResourceWithMainThread localResource,
ComponentMainThreadExecutor mainThreadExecutor) {
return (jobId, cleanupExecutor) ->
localResource.localCleanupAsync(jobId, cleanupExecutor, mainThreadExecutor);
}
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1686804638
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -164,7 +164,8 @@ void testFailingLocalCleanup() {
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ // Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.
Review Comment:
Done, thanks!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1686805757
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Done, LMK if you still want to collapse `DefaultJobManagerRunnerRegistry` and `OnMainThreadJobManagerRunnerRegistry`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Incorporated. LMK if you still want to collapse `DefaultJobManagerRunnerRegistry` and `OnMainThreadJobManagerRunnerRegistry`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1741026602
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
This might need to be slightly different since right now when this method is called, the result of the `closeAsync()`'s future is expected:
https://github.com/apache/flink/pull/25027/files#diff-780ffdbcb09b706ad230014f4b30772e393c67fdae253d9cf6a32adc188c5359R89-R95
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1888147138
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
The difference between your previous version of the code (using `thenApplyAsync`) and the changed code (using `thenRunAsync`) is a special (kind of annoying) issue with `CompletableFutures`:
The async `unregister` call is preceded by the [TestingJobManagerRunner#closeAsync](https://github.com/apache/flink/blob/153614716767d5fb8be1d4caa5e72167ce64a9d7/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java#L132) which is set up to be non-blocking. That means that the method's `CompletableFuture` is completed before returning.
There's a different runtime behavior between chained operations of an already-completed future and a not-yet-completed future. In the former case, the chained operation will be executed right-away on the calling thread . In the latter case, the chained operation will be scheduled for execution.
In our scenario (see [DefaultJobManagerRunnerRegistry#localCleanupAsync](https://github.com/apache/flink/blob/e37d9c9b7b649030d9cb5d3ba6540a006aa0aa23/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java#L91)), the `closeAsync` call returns the already-completed future. Keep in mind that this happens on the component's main thread! So, what happens with the chained operation?
* with `thenApplyAsync`: The `mainThreadExecutor#execute` call is called asynchronously (no executor is specified, so the JVM common thread pool is used). The `unregister` callback will be scheduled on the main thread again. But `localCleanupAsync` returns the future of the `thenApplyAsync` which completes as soon as the `mainThreadExecutor#execute` returns (not the `unregister` call!). That unblocks the "buggy" code I mentioned in [my previous comment](https://github.com/apache/flink/pull/25027#discussion_r1888127410).
* with `thenRunAsync`: We're on the main thread. `thenRunAsync` states that the `unregister` call should be executed on the main thread as well. But the `unregister` callback will be executed rightaway (instead of putting it on a task queue) due to the fact that the future that was returned by the previous `closeAsync` is already completed. In the mean time, the main thread is still blocked by the `ResourceCleaner` trying to complete the cleanup in the main thread. We reach a deadlock.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2231141258
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1686721979
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Ah I see! I was hesitant about this because that effectively means the `cleanupExecutor` for a `JobManagerRegistry` is always the `mainThreadExecutor`. But since `JobManagerRegistry`'s current implementations don't use the `cleanupExecutor` anyway so it's fine.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1872704374
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
I've been banging my head against this for a while now but can't figure this out. Do you have any thoughts @XComp ?
Using
```java
.thenRunAsync(() -> unregister(jobId), mainThreadExecutor);
```
sends the test `testNotArchivingSuspendedJobToHistoryServer` into a deadlock at line:
```java
dispatcher.getJobTerminationFuture(jobId, Duration.ofHours(1)).join();
```
But
```java
.thenRunAsync(() -> mainThreadExecutor.execute(() -> unregister(jobId)));
```
doesn't.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1828597077
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/OnMainThreadJobManagerRunnerRegistry.java:
##########
@@ -84,9 +84,10 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor executor) {
- mainThreadExecutor.assertRunningInMainThread();
- return delegate.localCleanupAsync(jobId, executor);
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, ComponentMainThreadExecutor mainThreadExecutor) {
+ this.mainThreadExecutor.assertRunningInMainThread();
Review Comment:
```suggestion
Preconditions.checkState(
mainThreadExecutor == this.mainThreadExecutor,
"A single main thread executor should be used.");
this.mainThreadExecutor.assertRunningInMainThread();
```
Let's make sure that we use the same main thread everywhere.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +84,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, ComponentMainThreadExecutor mainThreadExecutor) {
Review Comment:
```suggestion
JobID jobId, Executor ignoredCleanupExecutor, ComponentMainThreadExecutor mainThreadExecutor) {
```
nit: let's add the purpose to the parameter name here as well for better code readability.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableResourceWithMainThread.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableResourceWithMainThread} is an extension of the {@link
+ * LocallyCleanableResource} interface. It allows the {@link DispatcherResourceCleanerFactory} to
+ * inject the main thread as part of the cleanup procedure.
+ *
+ * <p>See {@code LocallyCleanableResource} for further context on the proper contract of the two
+ * interfaces.
+ *
+ * @see org.apache.flink.runtime.dispatcher.cleanup.LocallyCleanableResource
Review Comment:
```suggestion
```
nit: I guess, the link can be removed because we're linking to this interface in the text already.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,41 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableResourceWithMainThread>
Review Comment:
```suggestion
private final Collection<LocallyCleanableResourceWithMainThread>
```
The fields fields of the actual test instance are immutable
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -157,25 +144,33 @@ void testFailingLocalCleanup() {
assertThatFuture(
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor()))
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ new TestComponentMainThreadExecutor(Thread.currentThread())))
.isCompletedExceptionally()
.eventuallyFailsWith(ExecutionException.class)
.extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .as(
+ "Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.")
+ .isTrue();
}
@Test
- void testSuccessfulLocalCleanupAsync() {
Review Comment:
If we overload the method `registerTestingJobManagerRunner`:
```java
private TestingJobManagerRunner registerTestingJobManagerRunner() {
final TestingJobManagerRunner jobManagerRunner = TestingJobManagerRunner.newBuilder().build();
registerTestingJobManagerRunner(jobManagerRunner);
return jobManagerRunner;
}
private void registerTestingJobManagerRunner(
TestingJobManagerRunner jobManagerRunner) {
testInstance.register(jobManagerRunner);
assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isTrue();
assertThat(jobManagerRunner.getTerminationFuture()).isNotDone();
}
```
we can make this test even more reactive:
```java
@RegisterExtension
static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
TestingUtils.defaultExecutorExtension();
// [...]
@Test
void testSuccessfulLocalCleanupWaitsForDeregistration() {
final ManuallyTriggeredComponentMainThreadExecutor mainThreadExecutor =
new ManuallyTriggeredComponentMainThreadExecutor(Thread.currentThread());
final TestingJobManagerRunner jobManagerRunner = registerTestingJobManagerRunner();
final CompletableFuture<Void> cleanupResult =
testInstance.localCleanupAsync(
jobManagerRunner.getJobID(),
Executors.directExecutor(),
mainThreadExecutor);
jobManagerRunner.getTerminationFuture().complete(null);
FlinkAssertions.assertThatFuture(cleanupResult)
.as("Wait for the de-registration to complete")
.isNotDone();
mainThreadExecutor.triggerAll();
FlinkAssertions.assertThatFuture(cleanupResult).eventuallySucceeds();
}
@Test
void testSuccessfulLocalCleanupReliesOnJobManagerRunnerClose() {
final TestingJobManagerRunner jobManagerRunner =
TestingJobManagerRunner.newBuilder().setBlockingTermination(true).build();
registerTestingJobManagerRunner(jobManagerRunner);
final CompletableFuture<Void> cleanupResult =
CompletableFuture.runAsync(
() ->
testInstance.localCleanupAsync(
jobManagerRunner.getJobID(),
Executors.directExecutor(),
// execute de-registration right-away w/o any additional
// blocking
ComponentMainThreadExecutorServiceAdapter.forMainThread()),
// TestingJobManagerRunner doesn't close asynchronously, so we have to run
// the close call in a separate thread to unblock the test main thread
EXECUTOR_EXTENSION.getExecutor());
FlinkAssertions.assertThatFuture(cleanupResult)
.as("Wait for close call to complete")
.isNotDone();
jobManagerRunner.getTerminationFuture().complete(null);
FlinkAssertions.assertThatFuture(cleanupResult).eventuallySucceeds();
}
```
I didn't run the code. So it might need some adjustment.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestComponentMainThreadExecutor.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher;
+
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.operators.coordination.EventReceivingTasks;
+
+public class TestComponentMainThreadExecutor
Review Comment:
This is already available with [ComponentMainThreadExecutorServiceAdapter#forMainThread()](https://github.com/apache/flink/blob/b15a704b4f70590a6cefcf3bc6fcd412941717f9/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/ComponentMainThreadExecutorServiceAdapter.java#L60)
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -89,21 +103,30 @@ public static Builder builder() {
/** {@code Builder} for creating {@code TestingResourceCleanerFactory} instances. */
public static class Builder {
- private Collection<LocallyCleanableResource> locallyCleanableResources = new ArrayList<>();
+ private Collection<LocallyCleanableResource> locallyCleanableResource = new ArrayList<>();
+ private Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads = new ArrayList<>();
Review Comment:
```suggestion
locallyCleanableResourcesWithMainThread = new ArrayList<>();
```
nit: moving the `s` to the right place - there is only one main thread but possibly multiple resources.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/DispatcherResourceCleanerFactoryTest.java:
##########
@@ -96,7 +96,7 @@ private JobManagerRunnerRegistry createJobManagerRunnerRegistry() {
return TestingJobManagerRunnerRegistry.builder()
.withLocalCleanupAsyncFunction(
- (jobId, executor) -> {
+ (jobId, executor, mainThreadExecutor) -> {
Review Comment:
```suggestion
(jobId, cleanupExecutor, mainThreadExecutor) -> {
```
nit
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherCleanupITCase.java:
##########
@@ -220,7 +220,8 @@ public void testCleanupNotCancellable() throws Exception {
final JobManagerRunnerRegistry jobManagerRunnerRegistry =
TestingJobManagerRunnerRegistry.newSingleJobBuilder(jobManagerRunnerEntry)
.withLocalCleanupAsyncFunction(
- (actualJobId, executor) -> jobManagerRunnerCleanupFuture)
+ (actualJobId, executor, mainThreadExector) ->
Review Comment:
```suggestion
(actualJobId, cleanupExecutor, mainThreadExector) ->
```
nit: just to differentiate the two more explicitly
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -89,21 +103,30 @@ public static Builder builder() {
/** {@code Builder} for creating {@code TestingResourceCleanerFactory} instances. */
public static class Builder {
- private Collection<LocallyCleanableResource> locallyCleanableResources = new ArrayList<>();
+ private Collection<LocallyCleanableResource> locallyCleanableResource = new ArrayList<>();
+ private Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads = new ArrayList<>();
private Collection<GloballyCleanableResource> globallyCleanableResources =
new ArrayList<>();
private Executor cleanupExecutor = Executors.directExecutor();
- public Builder setLocallyCleanableResources(
- Collection<LocallyCleanableResource> locallyCleanableResources) {
- this.locallyCleanableResources = locallyCleanableResources;
+ public Builder setLocallyCleanableInMainThreadResource(
+ Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThread) {
+ this.locallyCleanableResource = locallyCleanableResource;
return this;
}
public Builder withLocallyCleanableResource(
LocallyCleanableResource locallyCleanableResource) {
- this.locallyCleanableResources.add(locallyCleanableResource);
+ this.locallyCleanableResource.add(locallyCleanableResource);
+ return this;
+ }
+
+ public Builder withLocallyCleanableInMainThreadResource(
Review Comment:
```suggestion
public Builder withLocallyCleanableResourceInMainThreadResource(
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -89,21 +103,30 @@ public static Builder builder() {
/** {@code Builder} for creating {@code TestingResourceCleanerFactory} instances. */
public static class Builder {
- private Collection<LocallyCleanableResource> locallyCleanableResources = new ArrayList<>();
+ private Collection<LocallyCleanableResource> locallyCleanableResource = new ArrayList<>();
Review Comment:
```suggestion
private Collection<LocallyCleanableResource> locallyCleanableResources = new ArrayList<>();
```
What's the purpose of renaming this field? Keeping the `s` also makes the diff of this PR smaller slightly smaller.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -89,21 +103,30 @@ public static Builder builder() {
/** {@code Builder} for creating {@code TestingResourceCleanerFactory} instances. */
public static class Builder {
- private Collection<LocallyCleanableResource> locallyCleanableResources = new ArrayList<>();
+ private Collection<LocallyCleanableResource> locallyCleanableResource = new ArrayList<>();
+ private Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads = new ArrayList<>();
private Collection<GloballyCleanableResource> globallyCleanableResources =
new ArrayList<>();
private Executor cleanupExecutor = Executors.directExecutor();
- public Builder setLocallyCleanableResources(
- Collection<LocallyCleanableResource> locallyCleanableResources) {
- this.locallyCleanableResources = locallyCleanableResources;
+ public Builder setLocallyCleanableInMainThreadResource(
Review Comment:
```suggestion
public Builder setLocallyCleanableResourcesInMainThreadResource(
```
nit: to keep the naming consistent
##########
flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/EventReceivingTasks.java:
##########
Review Comment:
That change is not necessary if we use `ComponentMainThreadExecutorServiceAdapter#forMainThread`.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,41 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads;
Review Comment:
```suggestion
locallyCleanableResourcesWithMainThreads;
```
nit
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingJobManagerRunnerRegistry.java:
##########
@@ -50,7 +51,8 @@ public class TestingJobManagerRunnerRegistry implements JobManagerRunnerRegistry
private final Supplier<Set<JobID>> getRunningJobIdsSupplier;
private final Supplier<Collection<JobManagerRunner>> getJobManagerRunnersSupplier;
private final Function<JobID, JobManagerRunner> unregisterFunction;
- private final BiFunction<JobID, Executor, CompletableFuture<Void>> localCleanupAsyncFunction;
+ private final TriFunction<JobID, Executor, Executor, CompletableFuture<Void>>
Review Comment:
```suggestion
private final TriFunction<JobID, Executor, ComponentMainThreadExecutor, CompletableFuture<Void>>
```
Let's make it match the actual signature to avoid running into problems down the road.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java:
##########
@@ -191,7 +206,7 @@ private TestingDispatcher.Builder createTestingDispatcherBuilder() {
// JobManagerRunnerRegistry needs to be added explicitly
// because cleaning it will trigger the closeAsync latch
// provided by TestingJobManagerRunner
- .withLocallyCleanableResource(jobManagerRunnerRegistry)
+ .withLocallyCleanableInMainThreadResource(jobManagerRunnerRegistry)
Review Comment:
Why are all the other changes in this class necessary? Isn't this change good enough? 🤔
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingJobManagerRunnerRegistry.java:
##########
@@ -188,10 +192,10 @@ public static class Builder {
private Supplier<Collection<JobManagerRunner>> getJobManagerRunnersSupplier =
Collections::emptyList;
private Function<JobID, JobManagerRunner> unregisterFunction = ignoredJobId -> null;
- private BiFunction<JobID, Executor, CompletableFuture<Void>> localCleanupAsyncFunction =
- (ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
- private BiFunction<JobID, Executor, CompletableFuture<Void>> globalCleanupAsyncFunction =
- (ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private TriFunction<JobID, Executor, Executor, CompletableFuture<Void>>
+ localCleanupAsyncFunction =
+ (ignoredJobId, ignoredExecutor, mainThreadExecutor) ->
Review Comment:
```suggestion
(ignoredJobId, ignoredCleanupExecutor, mainThreadExecutor) ->
```
nit
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,41 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads;
private final Collection<GloballyCleanableResource> globallyCleanableResources;
private final Executor cleanupExecutor;
private TestingResourceCleanerFactory(
Collection<LocallyCleanableResource> locallyCleanableResources,
+ Collection<LocallyCleanableResourceWithMainThread>
+ locallyCleanableResourceWithMainThreads,
Review Comment:
```suggestion
locallyCleanableResourcesWithMainThreads,
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -157,25 +144,33 @@ void testFailingLocalCleanup() {
assertThatFuture(
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor()))
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ new TestComponentMainThreadExecutor(Thread.currentThread())))
.isCompletedExceptionally()
.eventuallyFailsWith(ExecutionException.class)
.extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .as(
+ "Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.")
+ .isTrue();
}
@Test
- void testSuccessfulLocalCleanupAsync() {
+ void testSuccessfulLocalCleanupAsync() throws InterruptedException, ExecutionException {
Review Comment:
```suggestion
void testSuccessfulLocalCleanupAsync() {
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
The jobId is coming from the calling context (i.e. `localCleanupAsync`), not from the future. The future returns `Void` which is actually just `null` if I'm not mistaken. So, it doesn't add any value. You can go ahead and use `.thenRunAsync(() -> unregister(jobId), mainThreadExecutor);` here.
You might want to play around a bit with the `CompletableFuture` utility methods to get a better understanding of how these future chains behave.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1673562535
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
That question is answered by the [method's JavaDoc](https://github.com/apache/flink/blob/6cdb97c42b01a2eb3306784d240e16d202313ba2/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableResource.java#L46):
> @param cleanupExecutor The fallback executor for IO-heavy operations.
So no, unfortunately, it's not the main thread executor.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1703145088
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
+ */
+@FunctionalInterface
+public interface LocallyCleanableInMainThreadResource {
Review Comment:
I would suggest a slightly different name: `LocallyCleanableResourceWithMainThread`.
All cleanable resources are cleaned in the main thread (i.e. the interface method is called from the main thread). The difference with this interface is that the main thread reference is also passed in.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/OnMainThreadJobManagerRunnerRegistry.java:
##########
@@ -84,9 +84,10 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor executor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, Executor executor) {
Review Comment:
```suggestion
JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor) {
```
we should have the role of the two executors being reflected in the parameter names. Otherwise, it might become quite confusing.
I'm wondering whether we should even use `ComponentMainThreadExecutor` as a type for the main thread executor to make things clear. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/DispatcherResourceCleanerFactory.java:
##########
@@ -124,10 +127,39 @@ JOB_MANAGER_METRIC_GROUP_LABEL, ofLocalResource(jobManagerMetricGroup))
* state when we terminate globally.
*
* @param localResource Local resource that we want to clean during a global cleanup.
+ * @param mainThreadExecutor Main thread executor for cleanup.
* @return Globally cleanable resource.
*/
- private static GloballyCleanableResource ofLocalResource(
+ private static GloballyCleanableResource toGloballyCleanableResource(
+ LocallyCleanableInMainThreadResource localResource,
+ ComponentMainThreadExecutor mainThreadExecutor) {
+ return (jobId, cleanupExecutor) ->
+ localResource.localCleanupAsync(jobId, cleanupExecutor, mainThreadExecutor);
+ }
+
+ /**
+ * A simple wrapper for the resources that don't have any artifacts that can outlive the {@link
+ * org.apache.flink.runtime.dispatcher.Dispatcher}, but we still want to clean up their local
+ * state when we terminate globally.
+ *
+ * @param localResource Local resource that we want to clean during a global cleanup.
+ * @return Globally cleanable resource.
+ */
+ private static GloballyCleanableResource toGloballyCleanableResource(
LocallyCleanableResource localResource) {
return localResource::localCleanupAsync;
}
+
+ /**
+ * Converts a LocallyCleanableInMainThreadResource object to a LocallyCleanableResource object.
+ *
+ * @param localResource LocallyCleanableInMainThreadResource that we want to translate.
+ * @return A LocallyCleanableResource.
+ */
+ public static LocallyCleanableResource toLocallyCleanableResource(
Review Comment:
```suggestion
@VisibleForTesting
public static LocallyCleanableResource toLocallyCleanableResource(
```
If we really need this one in test code, we should mark it as `@VisibleForTesting`. But I'd argue that this is an implementation detail of the `DispatcherResourceCleanerFactory` which shouldn't be exposed. WDYT?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingJobManagerRunnerRegistry.java:
##########
@@ -188,11 +197,26 @@ public static class Builder {
private Supplier<Collection<JobManagerRunner>> getJobManagerRunnersSupplier =
Collections::emptyList;
private Function<JobID, JobManagerRunner> unregisterFunction = ignoredJobId -> null;
- private BiFunction<JobID, Executor, CompletableFuture<Void>> localCleanupAsyncFunction =
- (ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private TriFunction<JobID, Executor, Executor, CompletableFuture<Void>>
+ localCleanupAsyncFunction =
+ (ignoredJobId, ignoredExecutor, mainThreadExecutor) ->
+ FutureUtils.completedVoidFuture();
private BiFunction<JobID, Executor, CompletableFuture<Void>> globalCleanupAsyncFunction =
(ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
Review Comment:
```suggestion
```
I guess you could remove this one in a hotfix commit if you like. That seems to be some legacy code from when the JobManagerRunnerRegistry implemented both, the `LocallyCleanableResource` and the `GloballyCleanableResource` interfaces.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
Review Comment:
```suggestion
* {@code LocallyCleanableInMainThreadResource} is an extension of the {@link
* LocallyCleanableResource} interface. It allows the {@link DispatcherResourceCleanerFactory} to
* inject the main thread as part of the cleanup procedure. along the cleanup procedure.
*
* <p>See {@code LocallyCleanableResource} for further context on the proper contract of the two
* interfaces.
```
I would propose a more purpose-focused JavaDoc here. WDYT?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -157,25 +144,33 @@ void testFailingLocalCleanup() {
assertThatFuture(
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor()))
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor()))
.isCompletedExceptionally()
.eventuallyFailsWith(ExecutionException.class)
.extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isTrue()
+ .as(
+ "Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.");
}
@Test
- void testSuccessfulLocalCleanupAsync() {
+ void testSuccessfulLocalCleanupAsync() throws InterruptedException, ExecutionException {
final TestingJobManagerRunner jobManagerRunner = registerTestingJobManagerRunner();
-
final CompletableFuture<Void> cleanupResult =
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor());
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor());
+
+ // Wait for the unregister future to complete
+ cleanupResult.get();
Review Comment:
```suggestion
FlinkAssertions.assertThatFuture(cleanupFuture)
.as("Wait for the unregistration to complete")
.eventuallySucceeds();
```
nit: there are utility methods for asserting `CompletableFuture` instances
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/MiniDispatcherTest.java:
##########
@@ -369,7 +377,9 @@ private MiniDispatcher createMiniDispatcher(
// JobManagerRunnerRegistry needs to be added explicitly
// because cleaning it will trigger the closeAsync latch
// provided by TestingJobManagerRunner
- .withLocallyCleanableResource(jobManagerRunnerRegistry)
+ .withLocallyCleanableResource(
Review Comment:
Here we are inconsistent. The `TestingResourceCleanerFactory` has a dedicated `withLocallyCleanableInMainThreadResource` method for this purpose. Or am I missing something? :thinking:
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -202,23 +202,33 @@ void testFailingLocalCleanupAsync() {
}
@Test
- void testLocalCleanupAsyncNonBlocking() {
+ void testLocalCleanupAsyncNonBlocking() throws ExecutionException, InterruptedException {
final TestingJobManagerRunner jobManagerRunner =
TestingJobManagerRunner.newBuilder().setBlockingTermination(true).build();
testInstance.register(jobManagerRunner);
// this call shouldn't block
final CompletableFuture<Void> cleanupFuture =
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), UnsupportedOperationExecutor.INSTANCE);
-
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor());
+
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isTrue()
+ .as(
+ "Since the cleanup future hasn't completed yet, the JobManagerRunner is expected to not have been unregistered.");
assertThat(jobManagerRunner.getTerminationFuture()).isNotCompleted();
assertThat(cleanupFuture).isNotCompleted();
+ // Complete the closeAsync() future
jobManagerRunner.getTerminationFuture().complete(null);
- assertThat(cleanupFuture).isCompleted();
+ // Wait for the unregistration to complete
+ cleanupFuture.get();
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isFalse()
+ .as("The unregister future is complete now.");
Review Comment:
```suggestion
.as("The unregister future is complete now.")
.isFalse();
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
```suggestion
return get(jobId)
.closeAsync()
.thenRunAsync(() -> unregister(jobId), mainThreadExecutor);
```
We can shorten this part of the code. WDYT?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingJobManagerRunnerRegistry.java:
##########
@@ -188,11 +197,26 @@ public static class Builder {
private Supplier<Collection<JobManagerRunner>> getJobManagerRunnersSupplier =
Collections::emptyList;
private Function<JobID, JobManagerRunner> unregisterFunction = ignoredJobId -> null;
- private BiFunction<JobID, Executor, CompletableFuture<Void>> localCleanupAsyncFunction =
- (ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private TriFunction<JobID, Executor, Executor, CompletableFuture<Void>>
+ localCleanupAsyncFunction =
+ (ignoredJobId, ignoredExecutor, mainThreadExecutor) ->
+ FutureUtils.completedVoidFuture();
private BiFunction<JobID, Executor, CompletableFuture<Void>> globalCleanupAsyncFunction =
(ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private Builder fromDefaultJobManagerRunnerRegistry(
Review Comment:
What's the purpose of this? It looks like we're map each of the `DefaultJobManagerRunnerRegistry` methods to the `TestingJobManagerRunnerRegistry` callback generating actually a "new" `DefaultJobManagerRunnerRegistry` instance wrapped in a `TestingJobManagerRunnerRegistry`. We could just use `DefaultJobManagerRunnerRegistry`, instead. Don't you think? :thinking:
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -157,25 +144,33 @@ void testFailingLocalCleanup() {
assertThatFuture(
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor()))
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor()))
.isCompletedExceptionally()
.eventuallyFailsWith(ExecutionException.class)
.extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isTrue()
+ .as(
+ "Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.");
Review Comment:
```suggestion
.as(
"Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.")
.isTrue();
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/LocallyCleanableInMainThreadResource.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.dispatcher.cleanup;
+
+import org.apache.flink.api.common.JobID;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * {@code LocallyCleanableInMainThreadResource} is supposed to be implemented by any class that
+ * provides artifacts for a given job that need to be cleaned up in the main thread after the job
+ * reached a local terminal state. Local cleanups that needs to be triggered for a global terminal
+ * state as well, need to be implemented using the {@link GloballyCleanableResource}.
+ *
+ * <p>The {@link DispatcherResourceCleanerFactory} provides a workaround to trigger some {@code
+ * LocallyCleanableInMainThreadResource} as globally cleanable. FLINK-26175 is created to cover a
+ * refactoring and straighten things out.
+ *
+ * @see org.apache.flink.api.common.JobStatus
+ */
+@FunctionalInterface
+public interface LocallyCleanableInMainThreadResource {
+
+ /**
+ * {@code localCleanupAsync} is expected to be called from the main thread and uses the passed
+ * {@code mainThreadExecutor} for cleanups. Thread-safety must be ensured.
+ *
+ * @param jobId The {@link JobID} of the job for which the local data should be cleaned up.
+ * @param cleanupExecutor The fallback executor for IO-heavy operations.
+ * @param mainThreadExecutor The main thread executor.
+ * @return The cleanup result future.
+ */
+ CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
Review Comment:
```suggestion
JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
```
We could move the conversion also into the interface if we really need this in multiple places rather than only in the `[DispatcherResourceCleanerFactory`. WDYT?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -190,8 +185,13 @@ void testFailingLocalCleanupAsync() {
final CompletableFuture<Void> cleanupResult =
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), Executors.directExecutor());
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor());
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isTrue()
+ .as(
+ "Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.");
Review Comment:
```suggestion
.as(
"Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.")
.isTrue();
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,42 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources;
private final Collection<GloballyCleanableResource> globallyCleanableResources;
private final Executor cleanupExecutor;
private TestingResourceCleanerFactory(
Collection<LocallyCleanableResource> locallyCleanableResources,
+ Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources,
Collection<GloballyCleanableResource> globallyCleanableResources,
Executor cleanupExecutor) {
this.locallyCleanableResources = locallyCleanableResources;
+ this.locallyCleanableInMainThreadResources = locallyCleanableInMainThreadResources;
this.globallyCleanableResources = globallyCleanableResources;
this.cleanupExecutor = cleanupExecutor;
}
@Override
public ResourceCleaner createLocalResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor) {
+ Collection<LocallyCleanableResource> locallyCleanableResources =
+ this.locallyCleanableResources;
+
+ locallyCleanableResources.addAll(
+ locallyCleanableInMainThreadResources.stream()
+ .map(
+ r ->
+ DispatcherResourceCleanerFactory.toLocallyCleanableResource(
Review Comment:
`TestingResourceCleanerFactory` is a testing implementation that shouldn't depend on the `DispatcherResourceCleanerFactory`-internal code. We can do the mapping from `LocallyCleanableInMainThreadResource` to `LocallyCleanableResource` internally, I guess :thinking:
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -202,23 +202,33 @@ void testFailingLocalCleanupAsync() {
}
@Test
- void testLocalCleanupAsyncNonBlocking() {
+ void testLocalCleanupAsyncNonBlocking() throws ExecutionException, InterruptedException {
final TestingJobManagerRunner jobManagerRunner =
TestingJobManagerRunner.newBuilder().setBlockingTermination(true).build();
testInstance.register(jobManagerRunner);
// this call shouldn't block
final CompletableFuture<Void> cleanupFuture =
testInstance.localCleanupAsync(
- jobManagerRunner.getJobID(), UnsupportedOperationExecutor.INSTANCE);
-
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ jobManagerRunner.getJobID(),
+ Executors.directExecutor(),
+ Executors.directExecutor());
+
+ assertThat(testInstance.isRegistered(jobManagerRunner.getJobID()))
+ .isTrue()
+ .as(
+ "Since the cleanup future hasn't completed yet, the JobManagerRunner is expected to not have been unregistered.");
Review Comment:
```suggestion
.as(
"Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.")
.isTrue();
```
The assert message has to be put in front of the assertion.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1682297518
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
hm, what about a 4th option where we implement the `DispatcherResourceCleanerFactory` using the Builder pattern? That would allow us to set the `JobManagerRunnerRegistry` after instantiating the `ResourceCleanerFactory` instance. 🤔
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1679572415
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
Hi @XComp, I had to pass the `mainExecutor` to `DefaultJobManagerRegistry`. Can you please take a look if that makes sense?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1872707329
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
The Termination Future is acquired but `.join()` ends up waiting indefinitely. Also, I see the implementation of `.thenRunAsync(<task>, <executor>)` is different from just `<executor>.execute(<task>)`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771347067
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java:
##########
@@ -26,27 +26,42 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
/** {@code TestingResourceCleanerFactory} for adding custom {@link ResourceCleaner} creation. */
public class TestingResourceCleanerFactory implements ResourceCleanerFactory {
private final Collection<LocallyCleanableResource> locallyCleanableResources;
+ private Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources;
private final Collection<GloballyCleanableResource> globallyCleanableResources;
private final Executor cleanupExecutor;
private TestingResourceCleanerFactory(
Collection<LocallyCleanableResource> locallyCleanableResources,
+ Collection<LocallyCleanableInMainThreadResource> locallyCleanableInMainThreadResources,
Collection<GloballyCleanableResource> globallyCleanableResources,
Executor cleanupExecutor) {
this.locallyCleanableResources = locallyCleanableResources;
+ this.locallyCleanableInMainThreadResources = locallyCleanableInMainThreadResources;
this.globallyCleanableResources = globallyCleanableResources;
this.cleanupExecutor = cleanupExecutor;
}
@Override
public ResourceCleaner createLocalResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor) {
+ Collection<LocallyCleanableResource> locallyCleanableResources =
+ this.locallyCleanableResources;
+
+ locallyCleanableResources.addAll(
+ locallyCleanableInMainThreadResources.stream()
+ .map(
+ r ->
+ DispatcherResourceCleanerFactory.toLocallyCleanableResource(
Review Comment:
yes 👍
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1771603366
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/DispatcherResourceCleanerFactory.java:
##########
@@ -124,10 +127,39 @@ JOB_MANAGER_METRIC_GROUP_LABEL, ofLocalResource(jobManagerMetricGroup))
* state when we terminate globally.
*
* @param localResource Local resource that we want to clean during a global cleanup.
+ * @param mainThreadExecutor Main thread executor for cleanup.
* @return Globally cleanable resource.
*/
- private static GloballyCleanableResource ofLocalResource(
+ private static GloballyCleanableResource toGloballyCleanableResource(
+ LocallyCleanableInMainThreadResource localResource,
+ ComponentMainThreadExecutor mainThreadExecutor) {
+ return (jobId, cleanupExecutor) ->
+ localResource.localCleanupAsync(jobId, cleanupExecutor, mainThreadExecutor);
+ }
+
+ /**
+ * A simple wrapper for the resources that don't have any artifacts that can outlive the {@link
+ * org.apache.flink.runtime.dispatcher.Dispatcher}, but we still want to clean up their local
+ * state when we terminate globally.
+ *
+ * @param localResource Local resource that we want to clean during a global cleanup.
+ * @return Globally cleanable resource.
+ */
+ private static GloballyCleanableResource toGloballyCleanableResource(
LocallyCleanableResource localResource) {
return localResource::localCleanupAsync;
}
+
+ /**
+ * Converts a LocallyCleanableInMainThreadResource object to a LocallyCleanableResource object.
+ *
+ * @param localResource LocallyCleanableInMainThreadResource that we want to translate.
+ * @return A LocallyCleanableResource.
+ */
+ public static LocallyCleanableResource toLocallyCleanableResource(
Review Comment:
Removed this method in favour of a default method in `LocallyCleanableResourceWithMainThread`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
The main problem here seems to be the constructor chain we have in `Dispatcher`.
Here are the two non-private constructors:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
I see three approaches to do this:
1. [Initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor in say an `initDispatcher(...)` method - probably not a good idea since the attributes would need to be not `final`.
Instead of having the constructor chain, we would have two "base" constructors for `Dispatcher` only:
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
2. We can have two separate constructors which have redundant initializations of the attributes.
3. Constructor (I) would call (II) as:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices) {
this(
rpcService,
fencingToken,
recoveredJobs,
recoveredDirtyJobs,
dispatcherBootstrapFactory,
dispatcherServices,
null,
null);
}
```
In the base constructor, we'll initialize `jobManagerRunnerRegistry/resourceCleanerFactory` when they're null.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1703278478
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/TestingJobManagerRunnerRegistry.java:
##########
@@ -188,11 +197,26 @@ public static class Builder {
private Supplier<Collection<JobManagerRunner>> getJobManagerRunnersSupplier =
Collections::emptyList;
private Function<JobID, JobManagerRunner> unregisterFunction = ignoredJobId -> null;
- private BiFunction<JobID, Executor, CompletableFuture<Void>> localCleanupAsyncFunction =
- (ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private TriFunction<JobID, Executor, Executor, CompletableFuture<Void>>
+ localCleanupAsyncFunction =
+ (ignoredJobId, ignoredExecutor, mainThreadExecutor) ->
+ FutureUtils.completedVoidFuture();
private BiFunction<JobID, Executor, CompletableFuture<Void>> globalCleanupAsyncFunction =
(ignoredJobId, ignoredExecutor) -> FutureUtils.completedVoidFuture();
+ private Builder fromDefaultJobManagerRunnerRegistry(
Review Comment:
What's the purpose of this? It looks like we map each of the `DefaultJobManagerRunnerRegistry` methods to the `TestingJobManagerRunnerRegistry` callback generating actually a "new" `DefaultJobManagerRunnerRegistry` instance wrapped in a `TestingJobManagerRunnerRegistry`. We could just use `DefaultJobManagerRunnerRegistry`, instead. Don't you think? :thinking:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2781251887
This PR is being marked as stale since it has not had any activity in the last 90 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 30 days, it will be automatically closed.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1668120064
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
Review Comment:
What about using `thenAccept` here? ...since we're only interested in the success scenario.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
Here, we have to be careful: The `closeAsync` is not necessarily called in the `Dispatcher`'s main thread (let me know if you're unfamiliar with the main thread concept in Flink).
Essentially, the idea is for certain components to run all the logic in a single thread, i.e. the main thread, to avoid worrying about concurrency. That's the case for the `Dispatcher` (as it implements the `RPCEndpoint` interface) and all its sub-components. See [OnMainThreadJobManagerRunnerRegistry](https://github.com/apache/flink/blob/3efd4c2cf1be670f499e6637445e283e48deee60/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/OnMainThreadJobManagerRunnerRegistry.java) and its usages which ensures that we follow this paradigm for the `JobManagerRunnerRegistry` component.
I guess we would have to utilize `thenAcceptAsync` on the `closeAsync` future.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistryTest.java:
##########
@@ -164,7 +164,8 @@ void testFailingLocalCleanup() {
.hasExactlyElementsOfTypes(ExecutionException.class, expectedException.getClass())
.last()
.isEqualTo(expectedException);
- assertThat(testInstance.isRegistered(jobManagerRunner.getJobID())).isFalse();
+ // Since the cleanup failed, the JobManagerRunner is expected to not have been unregistered.
Review Comment:
nit: you could use the `.as()` method of JUnit5 to add a assert message rather than a comment. It would serve the same function as a comment but would also expose the description programmatically.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1668545338
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
I thought `thenAccept/whenComplete` would execute the action on the calling thread (the `closeAync` one) only and `thenAcceptAsync/whenCompleteAsync` on a separate thread.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1668650443
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -85,7 +85,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
@Override
public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ resultFuture.whenComplete(
+ (result, throwable) -> {
+ if (throwable == null) {
+ unregister(jobId);
Review Comment:
you're correct - that's what happens in the general scenario (there's an edge case where the `CompletableFuture` is already completed when chaining the callback to the future instance).
Executing the `unregister(JobID)` in the thread that executed `closeAsync` is something we don't want to do (for the reasons around main thread execution which I outlined in my comment above), though.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][dispatcher] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2211170007
@flinkbot run azure
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1888127410
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
Uh, that was a tricky one, I have to admit. The cause is that the [TestingResourceCleanerFactory#createResourceCleaner](https://github.com/apache/flink/blob/28791382c37b77a7c8db550a54e50fc50d591e35/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/cleanup/TestingResourceCleanerFactory.java#L81) provides a "buggy"/not-well-implemented `ResourceCleaner` implementation.
The `ResourceCleaner` is supposed to be called on the main thread. There shouldn't be any blocking functionality (because that would block other calls on the main thread). Instead, we should make the test implementation async as well:
```java
private <T> ResourceCleaner createResourceCleaner(
ComponentMainThreadExecutor mainThreadExecutor,
Collection<T> resources,
DefaultResourceCleaner.CleanupFn<T> cleanupFn) {
return jobId -> {
mainThreadExecutor.assertRunningInMainThread();
final Collection<CompletableFuture<Void>> asyncCleanupCallbackResults = resources
.stream()
.map(resource -> cleanupFn.cleanupAsync(resource, jobId, cleanupExecutor))
.collect(
Collectors.toList());
return FutureUtils.completeAll(asyncCleanupCallbackResults);
};
}
```
Changing the code to the one above fixes the issue with the `testNotArchivingSuspendedJobToHistoryServer` test.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,16 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(
+ JobID jobId, Executor ignoredExecutor, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApplyAsync(
+ result -> {
+ mainThreadExecutor.execute(() -> unregister(jobId));
+ return result;
+ });
Review Comment:
The difference between your previous version of the code (using `thenApplyAsync`) and the changed code (using `thenRunAsync`) is a special (kind of annoying) issue with `CompletableFutures`:
The async `unregister` call is preceded by the [TestingJobManagerRunner#closeAsync](https://github.com/apache/flink/blob/153614716767d5fb8be1d4caa5e72167ce64a9d7/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java#L132) which is set up to be non-blocking. That means that the method's `CompletableFuture` is completed before returning.
There's a different runtime behavior between chained operations of an already-completed future and a not-yet-completed future. In the former case, the chained operation will be executed right-away on the calling thread . In the latter case, the chained operation will be scheduled for execution.
In our scenario (see [DefaultJobManagerRunnerRegistry#localCleanupAsync](https://github.com/apache/flink/blob/e37d9c9b7b649030d9cb5d3ba6540a006aa0aa23/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java#L91)), the `closeAsync` call returns the already-completed future. Keep in mind that this happens on the component's main thread! So, what happens with the chained operation?
* with `thenApplyAsync`: The `mainThreadExecutor#execute` call is called asynchronously (no executor is specified, so the JVM common thread pool is used). The `unregister` callback will be scheduled on the main thread again. But `localCleanupAsync` returns the future of the `thenApplyAsync` which completes as soon as the `mainThreadExecutor#execute` returns (not the `unregister` call!). That unblocks the "buggy" code I mentioned in my previous comment.
* with `thenRunAsync`: We're on the main thread. `thenRunAsync` states that the `unregister` call should be executed on the main thread as well. But the `unregister` callback will be executed rightaway (instead of putting it on a task queue) due to the fact that the future that was returned by the previous `closeAsync` is already completed. In the mean time, the main thread is still blocked by the `ResourceCleaner` trying to complete the cleanup in the main thread. We reach a deadlock.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1692695511
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,15 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApply(
Review Comment:
I'm curious about this. Do we need to do `thenApplyAsync`, since the function just schedules a task and isn't blocking? 🤔
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on PR #25027:
URL: https://github.com/apache/flink/pull/25027#issuecomment-2261137905
@XComp , could you please take another look. I'll work on figuring out the following meanwhile:
> I'm wondering whether you could come up with a test where we verify that the close is called in another thread and the unregister is then executed in the main thread.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1692705494
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -83,9 +83,15 @@ public Collection<JobManagerRunner> getJobManagerRunners() {
}
@Override
- public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor unusedExecutor) {
+ public CompletableFuture<Void> localCleanupAsync(JobID jobId, Executor mainThreadExecutor) {
if (isRegistered(jobId)) {
- return unregister(jobId).closeAsync();
+ CompletableFuture<Void> resultFuture = this.jobManagerRunners.get(jobId).closeAsync();
+
+ return resultFuture.thenApply(
Review Comment:
One thing is that `thenApplyAsync` is the appropriate API to call a subsequent callback in a different thread (that's what you're currently doing with `mainThreadExecutor.execute`. Additionally, cleanup (i.e. the returned `CompletableFuture`) should be only completed after the `JobManagerRunner` is deregistered, IIUC. That's not the case in your current code where you trigger the unregister method but return the `result` right-away.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1686411653
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
JobManagerRunnerRegistry would implement `LocallyCleanableInMainThreadResource` instead of `LocallyCleanableResource`. The conversion of the `LocallyCleanableInMainThreadResource#localCleanupAsync` callback to the `LocallyCleanableResource#localCleanupAsync` callback would then happen in `DispatcherResourceCleanerFactory#create[Loc|Glob]alResourceCleaner`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
`JobManagerRunnerRegistry` would implement `LocallyCleanableInMainThreadResource` instead of `LocallyCleanableResource`. The conversion of the `LocallyCleanableInMainThreadResource#localCleanupAsync` callback to the `LocallyCleanableResource#localCleanupAsync` callback would then happen in `DispatcherResourceCleanerFactory#create[Loc|Glob]alResourceCleaner`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "kumar-mallikarjuna (via GitHub)" <gi...@apache.org>.
kumar-mallikarjuna commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1681117423
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea (since the attributes must be final)?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
Is [initializing these attributes](https://github.com/apache/flink/blob/583aadf97b2e3ddc87d4d244c5d62823e82513b1/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java#L272-L333) outside of the `Dispatcher` constructor a bad idea (since the attributes must be `final`)?
I'm thinking instead of having the constructor chain, we could have two "base" constructors for `Dispatcher` only:
I:
```java
public Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices)
```
and
II:
```java
protected Dispatcher(
RpcService rpcService,
DispatcherId fencingToken,
Collection<JobGraph> recoveredJobs,
Collection<JobResult> recoveredDirtyJobs,
DispatcherBootstrapFactory dispatcherBootstrapFactory,
DispatcherServices dispatcherServices,
JobManagerRunnerRegistry jobManagerRunnerRegistry,
ResourceCleanerFactory resourceCleanerFactory)
```
which call `super()` and a method `initDispatcher(...)` - to populate the attributes.
This way, we'll have a separate constructor for tests which can receive `jobManagerRunnerRegistry/resourceCleanerFactory` and we can build these two params in constructor (I).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-27355][runtime] Unregister JobManagerRunner after it's closed [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #25027:
URL: https://github.com/apache/flink/pull/25027#discussion_r1683903323
##########
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DefaultJobManagerRunnerRegistry.java:
##########
@@ -97,6 +106,14 @@ public JobManagerRunner unregister(JobID jobId) {
return this.jobManagerRunners.remove(jobId);
}
+ public void setMainThreadExecutor(ComponentMainThreadExecutor executor) {
+ mainThreadExecutor = executor;
+ }
+
+ public ComponentMainThreadExecutor getMainThreadExecutor() {
+ return mainThreadExecutor;
+ }
+
Review Comment:
yeah, I just noticed when thinking about it that using the Builder pattern doesn't make a difference. Essentially, it's about adding the `JobManagerRunnerRegistry` to the `create*ResourceCleaner` method signatures of the `ResourceCleanerFactory` which is a bit unfortunate.
But what we can do is making the main thread passing an implementation detail of the `DispatcherResourceCleanerFactory`: We could introduce a twin interface `LocallyCleanableWithMainThreadResource`:
```java
@FunctionalInterface
public interface LocallyCleanableInMainThreadResource {
CompletableFuture<Void> localCleanupAsync(
JobID jobId, Executor cleanupExecutor, Executor mainThreadExecutor);
}
```
That interface can be converted to a `LocallyCleanableResource` in `DispatcherResourceCleanerFactory`. We wouldn't have to change the `Dispatcher` instantiation in that case. 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org