You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by ab...@apache.org on 2026/03/23 09:45:12 UTC
(gobblin) branch master updated: [GOBBLIN-2257] Parallelize flow compilation on submission path (#4175)
This is an automated email from the ASF dual-hosted git repository.
abhijain pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 2613bb8a19 [GOBBLIN-2257] Parallelize flow compilation on submission path (#4175)
2613bb8a19 is described below
commit 2613bb8a19819c46eb17c756114c21f486a25e34
Author: Daisy Modi <da...@gmail.com>
AuthorDate: Mon Mar 23 15:15:04 2026 +0530
[GOBBLIN-2257] Parallelize flow compilation on submission path (#4175)
* Parallelize flow compilation on submission path
Flow compilation happens twice: on submission and on execution. The execution
path already runs on a thread pool of size 3 (DagProcessingEngine), but the
submission path was serialized by a synchronized block in
SpecCatalogListenersList.onAddSpec() and a single-thread executor in
CallbacksDispatcher.
This change removes the synchronized keyword from onAddSpec() and replaces
the single-thread executor with a configurable thread pool (default 3 threads),
allowing multiple flows to compile in parallel during submission. Compilation
is still serialized per flow but multiple flows now compile concurrently.
The thread pool size is configurable via:
gobblin.service.specCatalogListener.numThreads (default: 3)
Co-Authored-By: Claude Opus 4.6 (1M context) <no...@anthropic.com>
* Use bounded ThreadPoolExecutor with CallerRunsPolicy
Replace Executors.newFixedThreadPool (unbounded queue) with a bounded
ThreadPoolExecutor to prevent memory growth under heavy submission load.
Queue capacity is set to numThreads * 10 with CallerRunsPolicy so that
when the queue is full, the submitting thread runs the compilation
itself, providing natural backpressure.
Co-Authored-By: Claude Opus 4.6 (1M context) <no...@anthropic.com>
* Add comments explaining parallelization rationale
Add inline comment in FlowCatalog explaining why the submission path
is now parallelized and Javadoc on the new SpecCatalogListenersList
constructor documenting the design decision and thread-safety reasoning.
Co-Authored-By: Claude Opus 4.6 (1M context) <no...@anthropic.com>
---------
Co-authored-by: Daisy Modi <dm...@linkedin.com>
Co-authored-by: Claude Opus 4.6 (1M context) <no...@anthropic.com>
---
.../apache/gobblin/service/ServiceConfigKeys.java | 2 ++
.../gobblin/runtime/spec_catalog/FlowCatalog.java | 20 +++++++++++++++++++-
.../spec_catalog/SpecCatalogListenersList.java | 15 ++++++++++++++-
3 files changed, 35 insertions(+), 2 deletions(-)
diff --git a/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java b/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
index f18566c22f..bd526688d9 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/service/ServiceConfigKeys.java
@@ -159,6 +159,8 @@ public class ServiceConfigKeys {
public static final String NUM_DAG_PROC_THREADS_KEY = GOBBLIN_SERVICE_DAG_PROCESSING_ENGINE_PREFIX + "numThreads";
public static final String DAG_PROC_ENGINE_NON_RETRYABLE_EXCEPTIONS_KEY = GOBBLIN_SERVICE_DAG_PROCESSING_ENGINE_PREFIX + "nonRetryableExceptions";
public static final Integer DEFAULT_NUM_DAG_PROC_THREADS = 3;
+ public static final String NUM_SPEC_CATALOG_LISTENER_THREADS_KEY = GOBBLIN_SERVICE_PREFIX + "specCatalogListener.numThreads";
+ public static final int DEFAULT_NUM_SPEC_CATALOG_LISTENER_THREADS = 3;
public static final long DEFAULT_FLOW_FINISH_DEADLINE_MILLIS = TimeUnit.HOURS.toMillis(24);
public static final String ERROR_PATTERN_STORE_CLASS = ServiceConfigKeys.GOBBLIN_SERVICE_PREFIX + "errorPatternStore.class";
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java
index 24732cc862..8ac0214549 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java
@@ -27,6 +27,10 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.reflect.ConstructorUtils;
import org.slf4j.Logger;
@@ -62,6 +66,7 @@ import org.apache.gobblin.runtime.spec_store.FSSpecStore;
import org.apache.gobblin.service.ServiceConfigKeys;
import org.apache.gobblin.util.ClassAliasResolver;
import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.ExecutorsUtils;
import org.apache.gobblin.util.ExponentialBackoff;
import org.apache.gobblin.util.callbacks.CallbackResult;
import org.apache.gobblin.util.callbacks.CallbacksDispatcher;
@@ -110,7 +115,20 @@ public class FlowCatalog extends AbstractIdleService implements SpecCatalog, Mut
public FlowCatalog(Config config, Optional<Logger> log, Optional<MetricContext> parentMetricContext, boolean instrumentationEnabled) {
this.log = log.isPresent() ? log.get() : LoggerFactory.getLogger(getClass());
- this.listeners = new SpecCatalogListenersList(log);
+ // Flow compilation on the submission path was previously serialized by a synchronized block in
+ // SpecCatalogListenersList.onAddSpec() and a single-thread executor in CallbacksDispatcher.
+ // Since the execution path (DagProcessingEngine) already compiles flows concurrently on a thread pool
+ // of size 3, compileFlow() is proven thread-safe. We use a bounded ThreadPoolExecutor here to allow
+ // parallel compilation across flows during submission, with CallerRunsPolicy for backpressure under load.
+ int numListenerThreads = ConfigUtils.getInt(config,
+ ServiceConfigKeys.NUM_SPEC_CATALOG_LISTENER_THREADS_KEY,
+ ServiceConfigKeys.DEFAULT_NUM_SPEC_CATALOG_LISTENER_THREADS);
+ ExecutorService listenerExecutor = new ThreadPoolExecutor(numListenerThreads, numListenerThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>(numListenerThreads * 10),
+ ExecutorsUtils.newThreadFactory(log, Optional.of("SpecCatalogListenerThread-%d")),
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ this.listeners = new SpecCatalogListenersList(log, listenerExecutor);
if (instrumentationEnabled) {
MetricContext realParentCtx =
parentMetricContext.or(Instrumented.getMetricContext(new org.apache.gobblin.configuration.State(), getClass()));
diff --git a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/SpecCatalogListenersList.java b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/SpecCatalogListenersList.java
index e0c11aba48..12ae3ffe9c 100644
--- a/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/SpecCatalogListenersList.java
+++ b/gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/SpecCatalogListenersList.java
@@ -47,6 +47,19 @@ public class SpecCatalogListenersList implements SpecCatalogListener, SpecCatalo
_disp = new CallbacksDispatcher<>(Optional.<ExecutorService>absent(), log);
}
+ /**
+ * Constructor that accepts a custom {@link ExecutorService} for dispatching listener callbacks.
+ * This enables parallel flow compilation on the submission path. Previously, onAddSpec() was
+ * synchronized and used a single-thread executor, serializing all flow compilations. Since the
+ * execution path ({@link org.apache.gobblin.service.modules.orchestration.DagProcessingEngine})
+ * already runs compileFlow() from multiple threads, it is safe to parallelize here as well.
+ * Note: onDeleteSpec and onUpdateSpec remain synchronized as they handle infrequent TopologySpec
+ * mutations that modify shared state.
+ */
+ public SpecCatalogListenersList(Optional<Logger> log, ExecutorService executorService) {
+ _disp = new CallbacksDispatcher<>(Optional.of(executorService), log);
+ }
+
public Logger getLog() {
return _disp.getLog();
}
@@ -66,7 +79,7 @@ public class SpecCatalogListenersList implements SpecCatalogListener, SpecCatalo
}
@Override
- public synchronized AddSpecResponse onAddSpec(Spec addedSpec) {
+ public AddSpecResponse onAddSpec(Spec addedSpec) {
Preconditions.checkNotNull(addedSpec);
try {
return new AddSpecResponse<>(_disp.execCallbacks(new AddSpecCallback(addedSpec)));