Add dedicated analytics_scheduler thread pool to prevent deadlock

[Bukhtawar]

Introduce a dedicated analytics_scheduler thread pool (max(2, cpus/2) threads) for orchestration/wait. The reduce stage is dispatched on this scheduler pool, which immediately forks the actual reduce computation to SEARCH.
Fragment execution stays on SEARCH. The two layers can never starve each other.

Thread responsibilities after the change:

  ┌─────────────────────┬────────────────┬────────────────────────────────────────────────────────────────────────────────────┐
  │        Pool         │      Size      │                                        Role                                        │
  ├─────────────────────┼────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
  │ analytics_scheduler │ max(2, cpus/2) │ Orchestration/wait — holds during fragment completion, then forks reduce to SEARCH │
  ├─────────────────────┼────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
  │ SEARCH              │ (cpus×3)/2 + 1 │ All compute — fragment execution + reduce merge                                    │
  ├─────────────────────┼────────────────┼────────────────────────────────────────────────────────────────────────────────────┤
  │ Transport           │ (Netty I/O)    │ Delivers callbacks, never blocked                                                  │
  └─────────────────────┴────────────────┴────────────────────────────────────────────────────────────────────────────────────┘

Changes:

• AnalyticsPlugin: register analytics_coordinator FixedExecutorBuilder
• QueryContext: accept ThreadPool instead of individual Executors, expose searchExecutor() and coordinatorExecutor() accessors
• ReduceStageExecution/PassThroughStageExecution: use coordinatorExecutor()
• DefaultPlanExecutor: pass ThreadPool to QueryContext

Description

[Describe what this change achieves]

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 7200ad7)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

📝 TODO sections sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java [79]: close on shutdown — arrow-base's root.close() warns about outstanding child

🔀 No multiple PR themes

⚡ Recommended focus areas for review Exception Handling The reduce task wraps backendSink.reduce(listener) in a try-catch that calls listener.onFailure(e) on exception. If backendSink.reduce itself calls listener.onFailure before throwing, the listener receives two failure notifications. This can cause downstream code to process the same error twice or trigger unexpected state transitions. searchExecutor.execute(() -> { try { backendSink.reduce(listener); } catch (Exception e) { listener.onFailure(e); } Possible Issue schedulerPoolSize() returns max(2, cpus/2). On a single-CPU system, availableProcessors() returns 1, so cpus/2 is 0, and the pool size becomes max(2, 0) = 2. However, the PR description states the pool size is max(2, cpus/2) threads, implying the intent is to scale with CPU count. On systems with 2 or 3 CPUs, the pool size will be 2, which may be insufficient if the workload expects scaling. Verify whether this is intentional or if the formula should be max(2, cpus/2) with a minimum threshold that scales better on low-CPU systems. static int schedulerPoolSize() { return Math.max(2, Runtime.getRuntime().availableProcessors() / 2); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 7200ad7

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Handle executor rejection exceptions The searchExecutor.execute() call may throw RejectedExecutionException if the executor is shut down or the queue is full. This exception would propagate to the scheduler thread uncaught. Wrap the execute call in a try-catch to handle rejection gracefully and notify the listener. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/stage/coordinator/ReduceStageExecution.java [83-91] return List.of(new LocalStageTask(new StageTaskId(getStageId(), 0), listener -> { - searchExecutor.execute(() -> { - try { - backendSink.reduce(listener); - } catch (Exception e) { - listener.onFailure(e); - } - }); + try { + searchExecutor.execute(() -> { + try { + backendSink.reduce(listener); + } catch (Exception e) { + listener.onFailure(e); + } + }); + } catch (Exception e) { + listener.onFailure(e); + } })); Suggestion importance[1-10]: 7 __ Why: This is a valid concern about RejectedExecutionException when searchExecutor.execute() is called. The suggestion correctly identifies that exceptions from the executor submission itself are not caught, which could leave the listener without proper failure notification. This improves error handling robustness.	Medium
	Prevent NPE from null executors Both executor methods return inline execution (Runnable::run) when threadPool is null. If threadPool.executor() returns null (possible in shutdown scenarios), a NullPointerException will occur. Add null-checks for the executor return values to prevent NPE during edge cases. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/QueryContext.java [78-84] public Executor searchExecutor() { - return threadPool != null ? threadPool.executor(ThreadPool.Names.SEARCH) : Runnable::run; + if (threadPool == null) return Runnable::run; + Executor executor = threadPool.executor(ThreadPool.Names.SEARCH); + return executor != null ? executor : Runnable::run; } public Executor schedulerExecutor() { - return threadPool != null ? threadPool.executor(AnalyticsPlugin.SCHEDULER_THREAD_POOL_NAME) : Runnable::run; + if (threadPool == null) return Runnable::run; + Executor executor = threadPool.executor(AnalyticsPlugin.SCHEDULER_THREAD_POOL_NAME); + return executor != null ? executor : Runnable::run; } Suggestion importance[1-10]: 6 __ Why: The suggestion adds defensive null-checking for the case where threadPool.executor() might return null during shutdown or edge cases. While the current code handles threadPool being null, it doesn't guard against executor() returning null. This is a reasonable defensive programming improvement, though the likelihood of this scenario depends on the ThreadPool implementation.	Low
General	Validate processor count edge cases The schedulerPoolSize() method could return 0 when availableProcessors() returns 1, causing Math.max(2, 1/2) to evaluate to Math.max(2, 0) = 2. However, on systems reporting 0 processors (edge case), this would still return 2. Consider adding validation or documentation for minimum processor requirements. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java [158-160] static int schedulerPoolSize() { - return Math.max(2, Runtime.getRuntime().availableProcessors() / 2); + int processors = Runtime.getRuntime().availableProcessors(); + if (processors <= 0) { + return 2; // fallback for edge cases + } + return Math.max(2, processors / 2); } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses an edge case where availableProcessors() might return 0 or negative values. However, this is extremely rare in practice, and the current implementation already handles the common case (1 processor) correctly by returning 2. The added validation provides marginal defensive programming benefit.	Low

---

Previous suggestions

Suggestions up to commit 3522e69

Category	Suggestion	Impact
Possible issue	Handle executor rejection in reduce stage The lambda captures listener and passes it to backendSink.reduce(), but if searchExecutor.execute() throws a RejectedExecutionException (e.g., queue full), the listener will never be notified. Wrap the execute() call in a try-catch to ensure listener.onFailure() is called on rejection. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/stage/coordinator/ReduceStageExecution.java [83-91] return List.of(new LocalStageTask(new StageTaskId(getStageId(), 0), listener -> { - searchExecutor.execute(() -> { - try { - backendSink.reduce(listener); - } catch (Exception e) { - listener.onFailure(e); - } - }); + try { + searchExecutor.execute(() -> { + try { + backendSink.reduce(listener); + } catch (Exception e) { + listener.onFailure(e); + } + }); + } catch (Exception e) { + listener.onFailure(e); + } })); Suggestion importance[1-10]: 7 __ Why: This is a valid error handling improvement. If searchExecutor.execute() throws RejectedExecutionException, the listener would not be notified, leaving the caller hanging. Adding the outer try-catch ensures proper error propagation in all failure scenarios.	Medium
General	Increase minimum scheduler thread pool size The scheduler pool size calculation may result in only 2 threads on systems with 2-4 cores, which could become a bottleneck. Consider using a minimum of 4 threads or a different scaling formula to ensure adequate parallelism for query scheduling operations. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java [158-160] static int schedulerPoolSize() { - return Math.max(2, Runtime.getRuntime().availableProcessors() / 2); + return Math.max(4, Runtime.getRuntime().availableProcessors() / 2); } Suggestion importance[1-10]: 4 __ Why: The suggestion to increase the minimum from 2 to 4 threads is a reasonable performance optimization, but lacks concrete evidence that 2 threads would be a bottleneck. The impact is moderate as it's a configuration tuning suggestion without addressing a critical issue.	Low

Suggestions up to commit dbe4099

Category	Suggestion	Impact
Possible issue	Handle executor submission failures The searchExecutor.execute() call may fail if the executor is shut down or the queue is full, but this failure is not caught. This could leave the listener in a pending state indefinitely. Wrap the execute() call in a try-catch block and invoke listener.onFailure() if submission fails. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/stage/coordinator/ReduceStageExecution.java [83-91] return List.of(new LocalStageTask(new StageTaskId(getStageId(), 0), listener -> { - searchExecutor.execute(() -> { - try { - backendSink.reduce(listener); - } catch (Exception e) { - listener.onFailure(e); - } - }); + try { + searchExecutor.execute(() -> { + try { + backendSink.reduce(listener); + } catch (Exception e) { + listener.onFailure(e); + } + }); + } catch (Exception e) { + listener.onFailure(e); + } })); Suggestion importance[1-10]: 7 __ Why: Valid concern about executor submission failures leaving the listener in a pending state. However, this is error handling for an edge case (executor shutdown/queue full) rather than a critical bug in normal operation.	Medium
General	Increase minimum scheduler thread count The scheduler pool size calculation may return too few threads on systems with 2-3 processors (e.g., 2 CPUs → 1 thread → max(2,1)=2). Consider using a higher minimum threshold (e.g., 4) or a different formula to ensure adequate parallelism for query coordination tasks. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/AnalyticsPlugin.java [158-160] static int schedulerPoolSize() { - return Math.max(2, Runtime.getRuntime().availableProcessors() / 2); + return Math.max(4, Runtime.getRuntime().availableProcessors() / 2); } Suggestion importance[1-10]: 4 __ Why: The suggestion to increase the minimum from 2 to 4 threads is a minor optimization. The current formula already ensures at least 2 threads, which may be sufficient for the scheduler's coordination tasks. The impact is marginal without evidence of inadequate parallelism.	Low

[github-actions]

Persistent review updated to latest commit 3522e69

[github-actions]

❌ Gradle check result for 3522e69: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 3522e69: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

❌ Gradle check result for 3522e69: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 7200ad7

[github-actions]

❌ Gradle check result for 7200ad7: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

✅ Gradle check result for 7200ad7: SUCCESS

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.36%. Comparing base (ca22376) to head (7200ad7).
⚠️ Report is 1 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21771      +/-   ##
============================================
- Coverage     73.44%   73.36%   -0.09%     
+ Complexity    75084    75033      -51     
============================================
  Files          6016     6016              
  Lines        341072   341072              
  Branches      49091    49091              
============================================
- Hits         250513   250216     -297     
- Misses        70641    70884     +243     
- Partials      19918    19972      +54     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.