You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "becketqin (via GitHub)" <gi...@apache.org> on 2024/10/23 16:34:37 UTC
[PR] [FLINK-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
becketqin opened a new pull request, #25569:
URL: https://github.com/apache/flink/pull/25569
## What is the purpose of the change
This patch fixes an issue introduced in #25130. In `SplitFetcherManager.close()`, the element queue draining thread was chaining the runnables to the element queue availability future in a tight loop. This causes problem (e.g. OOM, high CPU util) when the fetcher threads do not shutdown quickly.
This patch changes the tight async loop to a blocking loop.
## Brief change log
This patch changes the tight async loop to a blocking loop.
## Verifying this change
This change added a unit test. However, that unit test is kind of testing against the implementation instead of the behavior. But given we are fixing an interal impl issue, this might be necessary.
## 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: (no)
- 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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin merged PR #25569:
URL: https://github.com/apache/flink/pull/25569
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on PR #25569:
URL: https://github.com/apache/flink/pull/25569#issuecomment-2669853890
I see the unit test to be flaky. Will fix it with a follow up hotfix patch.
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on code in PR #25569:
URL: https://github.com/apache/flink/pull/25569#discussion_r1866671173
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -100,6 +102,50 @@ void testCloseCleansUpPreviouslyClosedFetcher() throws Exception {
fetcherManager.close(Long.MAX_VALUE);
}
+ /**
Review Comment:
It might be better to keep this comment here. In general, the tests should be testing against the behavior. But in this particular case, the test is against the implementation. Keeping the comment helps the maintainers know that this test can actually be removed if the implementation changes.
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "tomncooper (via GitHub)" <gi...@apache.org>.
tomncooper commented on PR #25569:
URL: https://github.com/apache/flink/pull/25569#issuecomment-2593700395
Looks like this PR is good to go.
@becketqin can you run `mvn spotless:apply` and rebase to see if CI will pass?
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "vahmed-hamdy (via GitHub)" <gi...@apache.org>.
vahmed-hamdy commented on code in PR #25569:
URL: https://github.com/apache/flink/pull/25569#discussion_r1856326806
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -269,19 +270,30 @@ public synchronized void close(long timeoutMs) throws Exception {
// fetcher threads blocking on putting batches into the element queue.
executors.submit(
() -> {
- while (fetchersToShutDown.get() > 0
- && System.currentTimeMillis() - startTime < timeoutMs) {
- elementsQueue
- .getAvailabilityFuture()
- .thenRun(() -> elementsQueue.poll().recycle());
+ long timeElapsed = System.currentTimeMillis() - startTime;
+ while (fetchersToShutDown.get() > 0 && timeElapsed < timeoutMs) {
+ try {
+ elementsQueue
+ .getAvailabilityFuture()
+ .thenRun(() -> elementsQueue.poll().recycle())
+ .get(timeoutMs - timeElapsed, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ LOG.warn(
+ "Received exception when waiting for the fetchers to "
+ + "shutdown.",
+ e);
+ break;
+ }
+ timeElapsed = System.currentTimeMillis() - startTime;
}
});
executors.shutdown();
- if (!executors.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
+ long timeElapsed = System.currentTimeMillis() - startTime;
+ if (!executors.awaitTermination(timeoutMs - timeElapsed, TimeUnit.NANOSECONDS)) {
LOG.warn(
- "Failed to close the split fetchers in {} ms. There are still {} split fetchers running",
+ "Failed to close the source reader in {} ms. There are still {} split fetchers running",
Review Comment:
nit: why are we changing the warn message?
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -269,19 +270,30 @@ public synchronized void close(long timeoutMs) throws Exception {
// fetcher threads blocking on putting batches into the element queue.
executors.submit(
() -> {
- while (fetchersToShutDown.get() > 0
- && System.currentTimeMillis() - startTime < timeoutMs) {
- elementsQueue
- .getAvailabilityFuture()
- .thenRun(() -> elementsQueue.poll().recycle());
+ long timeElapsed = System.currentTimeMillis() - startTime;
+ while (fetchersToShutDown.get() > 0 && timeElapsed < timeoutMs) {
+ try {
+ elementsQueue
+ .getAvailabilityFuture()
+ .thenRun(() -> elementsQueue.poll().recycle())
+ .get(timeoutMs - timeElapsed, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
Review Comment:
Should we catch more specific? maybe we need to break on `InterruptedException` but we just continue trying to wait on the others if it is an `ExecutionException`?
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##########
@@ -100,6 +102,50 @@ void testCloseCleansUpPreviouslyClosedFetcher() throws Exception {
fetcherManager.close(Long.MAX_VALUE);
}
+ /**
Review Comment:
This comment could be removed
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "becketqin (via GitHub)" <gi...@apache.org>.
becketqin commented on PR #25569:
URL: https://github.com/apache/flink/pull/25569#issuecomment-2666553880
@PatrickRen @tomncooper Thanks for the review and sorry for the late reply. I just returned to work from vacation. I updated the patch to fix the checkstyle failure. Will merge the patch after the CI passes.
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on PR #25569:
URL: https://github.com/apache/flink/pull/25569#issuecomment-2678967540
@becketqin I'm very sorry however I had to revert it since after this PR was merged master hasn't been able to reach green CI for almost a week, links to failed CI are at FLINK-37373
reverted at https://github.com/apache/flink/commit/77b03d94ab3857f80379ce6a6d10c142bddf72e5
--
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-36585][connector / common] SplitFetcherManager.close() should not chain runnables to the element queue future in a tight loop. [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25569:
URL: https://github.com/apache/flink/pull/25569#issuecomment-2432821612
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "2c080d9096af72fba4da080e58b61314c13f62d4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2c080d9096af72fba4da080e58b61314c13f62d4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2c080d9096af72fba4da080e58b61314c13f62d4 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