You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2018/11/12 17:12:34 UTC
[beam] branch master updated: [BEAM-6018] Don't use
getExitingExecutorService (#6994)
This is an automated email from the ASF dual-hosted git repository.
lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new a8458c0 [BEAM-6018] Don't use getExitingExecutorService (#6994)
a8458c0 is described below
commit a8458c030f2e5c19507e01f8d325c0efef0162f9
Author: Udi Meiri <ud...@users.noreply.github.com>
AuthorDate: Mon Nov 12 09:12:27 2018 -0800
[BEAM-6018] Don't use getExitingExecutorService (#6994)
In GCSUtil, this causes a slow memory leak in streaming pipelines that
do windowed writes to GCS.
---
.../src/main/java/org/apache/beam/sdk/util/GcsUtil.java | 15 ++++++++-------
1 file changed, 8 insertions(+), 7 deletions(-)
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 4fa84fa..94f880e 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 class GcsUtil {
// 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 class GcsUtil {
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) {