[
https://issues.apache.org/jira/browse/BEAM-6018?focusedWorklogId=165019&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-165019
]
ASF GitHub Bot logged work on BEAM-6018:
----------------------------------------
Author: ASF GitHub Bot
Created on: 12/Nov/18 17:12
Start Date: 12/Nov/18 17:12
Worklog Time Spent: 10m
Work Description: lukecwik closed pull request #6994: [BEAM-6018] Fix
slow memory leak in GcsUtil
URL: https://github.com/apache/beam/pull/6994
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsUtil.java
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsUtil.java
index 4fa84fa22cc..94f880eebe1 100644
---
a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsUtil.java
+++
b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsUtil.java
@@ -136,6 +136,8 @@ public static GcsUtil create(
// Helper delegate for turning IOExceptions from API calls into higher-level
semantics.
private final ApiErrorExtractor errorExtractor = new ApiErrorExtractor();
+ // Unbounded thread pool for codependent pipeline operations that will
deadlock the pipeline if
+ // starved for threads.
// Exposed for testing.
final ExecutorService executorService;
@@ -548,13 +550,12 @@ public boolean shouldRetry(IOException e) {
private static void executeBatches(List<BatchRequest> batches) throws
IOException {
ExecutorService executor =
MoreExecutors.listeningDecorator(
- MoreExecutors.getExitingExecutorService(
- new ThreadPoolExecutor(
- MAX_CONCURRENT_BATCHES,
- MAX_CONCURRENT_BATCHES,
- 0L,
- TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<>())));
+ new ThreadPoolExecutor(
+ MAX_CONCURRENT_BATCHES,
+ MAX_CONCURRENT_BATCHES,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>()));
List<CompletionStage<Void>> futures = new ArrayList<>();
for (final BatchRequest batch : batches) {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 165019)
Time Spent: 1.5h (was: 1h 20m)
> Memory leak in GCSUtil.java executeBatches
> ------------------------------------------
>
> Key: BEAM-6018
> URL: https://issues.apache.org/jira/browse/BEAM-6018
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
> Affects Versions: 2.7.0
> Reporter: Udi Meiri
> Assignee: Udi Meiri
> Priority: Major
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> In streaming pipelines there are multiple calls to moveToOutputFiles
> (https://github.com/apache/beam/blob/42984a821b3e73aee2966d11d7fb436b5ff22b68/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L796).
> When writing to GCS, this call uses executeBatches
> (https://github.com/apache/beam/blob/279a05604b83a54e8e5a79e13d8761f94841f326/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/util/GcsUtil.java#L551),
> which wraps a thread pool in MoreExecutors.getExitingExecutorService(). This
> wrapper introduces a DelayedShutdownHook which persists until the worker
> exits.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
