[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) <noreply@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) <noreply@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) <noreply@anthropic.com>

---------

Co-authored-by: Daisy Modi <dmodi@linkedin.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

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";

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java

@@ -27,6 +27,10 @@
 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.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 FlowCatalog(Config config, GobblinInstanceEnvironment env) {
 
   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()));

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/SpecCatalogListenersList.java

@@ -47,6 +47,19 @@ public SpecCatalogListenersList(Optional<Logger> log) {
     _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 synchronized void removeListener(SpecCatalogListener oldListener) {
   }
 
   @Override
-  public synchronized AddSpecResponse onAddSpec(Spec addedSpec) {
+  public AddSpecResponse onAddSpec(Spec addedSpec) {
     Preconditions.checkNotNull(addedSpec);
     try {
       return new AddSpecResponse<>(_disp.execCallbacks(new AddSpecCallback(addedSpec)));