Add indexing throttle on merge pressure for DataFormatAwareEngine

[Shailesh-Kumar-Singh]

When outstanding merges (active + pending) exceed maxMergeCount, the MergeScheduler now activates indexing throttle via the engine's existing IndexingThrottler, serializing write threads to a single thread until merge pressure subsides. This mirrors the behavior already present in InternalEngine's EngineMergeScheduler.

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 a969638)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Throttle stuck at boundary In evaluateThrottle(), the condition uses numMergesInFlight > maxMergeCount to activate and numMergesInFlight < maxMergeCount to deactivate. When numMergesInFlight == maxMergeCount, neither branch executes, so if throttling was previously activated (e.g., went from maxMergeCount+1 down to maxMergeCount), it will not be deactivated until the count drops strictly below maxMergeCount. Consider using <= for the deactivate condition (mirroring InternalEngine semantics) so the throttle reliably clears. private synchronized void evaluateThrottle() { int numMergesInFlight = activeMerges.get() + mergeHandler.getPendingMergeCount(); if (numMergesInFlight > maxMergeCount) { if (isThrottling.getAndSet(true) == false) { logger.info("now throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { activateThrottling.run(); } catch (Exception e) { logger.warn("exception in activateThrottling callback", e); } } } else if (numMergesInFlight < maxMergeCount) { if (isThrottling.getAndSet(false)) { logger.info("stop throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { deactivateThrottling.run(); } catch (Exception e) { logger.warn("exception in deactivateThrottling callback", e); } } } } Synchronized callback risk evaluateThrottle() is synchronized and invokes the activateThrottling/deactivateThrottling callbacks while holding the monitor. If those callbacks (engine throttling logic) acquire other locks that may, on another path, call back into the scheduler's synchronized methods, this could lead to lock-order inversion / deadlock. Consider releasing the monitor before invoking external callbacks (e.g., capture the desired state under lock, then run the callback outside). private synchronized void evaluateThrottle() { int numMergesInFlight = activeMerges.get() + mergeHandler.getPendingMergeCount(); if (numMergesInFlight > maxMergeCount) { if (isThrottling.getAndSet(true) == false) { logger.info("now throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { activateThrottling.run(); } catch (Exception e) { logger.warn("exception in activateThrottling callback", e); } } } else if (numMergesInFlight < maxMergeCount) { if (isThrottling.getAndSet(false)) { logger.info("stop throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { deactivateThrottling.run(); } catch (Exception e) { logger.warn("exception in deactivateThrottling callback", e); } } } }

[github-actions]

Persistent review updated to latest commit af9a3f0

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to a969638
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix boundary handling in throttle thresholds The throttle activation condition numMergesInFlight > maxMergeCount and deactivation numMergesInFlight < maxMergeCount leaves the boundary == maxMergeCount unhandled, which means once throttling is on it stays on at the boundary and once off it stays off. Lucene's ConcurrentMergeScheduler activates throttling when >= maxMergeCount. Consider using >= for activation and < for deactivation (with hysteresis) to align with expected semantics. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [411-421] + if (numMergesInFlight >= maxMergeCount) { if (isThrottling.getAndSet(true) == false) { logger.info("now throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { activateThrottling.run(); } catch (Exception e) { logger.warn("exception in activateThrottling callback", e); } } - } else if (numMergesInFlight < maxMergeCount) { + } else { if (isThrottling.getAndSet(false)) { Suggestion importance[1-10]: 7 __ Why: Valid observation: at numMergesInFlight == maxMergeCount, neither branch executes, leaving throttle state stale. Aligning with Lucene's >= activation semantics is a reasonable correctness improvement, though impact is minor since it only affects the boundary case.	Medium
General	Avoid throttle eval blocking drain listeners When a merge throws, activeMerges is decremented but evaluateThrottle() runs synchronously on the merge thread; if deactivateThrottling.run() throws despite the try/catch around it, the drain listener logic below is skipped. Move evaluateThrottle() after the drain-listener block (or wrap it defensively) so listener firing is not affected by throttle callback issues. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [355-356] - } catch (Exception e) { - // runMerge already invoked onMergeFailureCleanup; swallow to prevent - // uncaught exception on the merge thread pool. } finally { activeMerges.decrementAndGet(); + // Fire all drain listeners if all merges completed and none pending + if (isFrozen() && activeMerges.get() == 0 && !mergeHandler.hasPendingMerges() && !onDrainedListeners.isEmpty()) { + // existing drain logic + } evaluateThrottle(); Suggestion importance[1-10]: 3 __ Why: The evaluateThrottle() method already wraps throttling callbacks in try/catch, so the risk of skipping drain listener logic is low. The suggestion offers marginal defensive-programming value.	Low

---

Previous suggestions

Suggestions up to commit af9a3f0

Category	Suggestion	Impact
Possible issue	Fix off-by-one in throttle deactivation The deactivation condition numMergesInFlight < maxMergeCount leaves a gap when numMergesInFlight == maxMergeCount, meaning throttling will not be deactivated if it equals the threshold (since activation requires strictly greater). Use <= for deactivation to ensure throttling is properly released once the in-flight count drops to or below the limit. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [420] private synchronized void evaluateThrottle() { int numMergesInFlight = activeMerges.get() + mergeHandler.getPendingMergeCount(); if (numMergesInFlight > maxMergeCount) { if (isThrottling.getAndSet(true) == false) { logger.info("now throttling indexing: numMergesInFlight={}, maxMergeCount={}", numMergesInFlight, maxMergeCount); try { activateThrottling.run(); } catch (Exception e) { logger.warn("exception in activateThrottling callback", e); } } - } else if (numMergesInFlight < maxMergeCount) { + } else if (numMergesInFlight <= maxMergeCount) { Suggestion importance[1-10]: 6 __ Why: Valid observation: when numMergesInFlight == maxMergeCount, neither branch executes, so throttling state may not be deactivated at the boundary. Using <= would correctly close the gap, though the practical impact is limited since the count typically continues decreasing.	Low
General	Skip throttle evaluation during shutdown evaluateThrottle() is invoked from triggerMerges() even when the scheduler is shutdown path is taken earlier, but if activateThrottling/deactivateThrottling callbacks invoke engine methods after the engine is closed, this could throw. Consider guarding evaluateThrottle against isShutdown to avoid invoking engine callbacks during/after shutdown. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [150] if (!isFrozen()) { mergeHandler.findAndRegisterMerges(); } -evaluateThrottle(); +if (!isShutdown.get()) { + evaluateThrottle(); +} executeMerge(); Suggestion importance[1-10]: 3 __ Why: The suggestion is speculative; the callbacks are already wrapped in try/catch in evaluateThrottle, and there's no clear evidence of an actual shutdown-related issue. Minor defensive improvement at best.	Low

[github-actions]

Persistent review updated to latest commit a969638

[github-actions]

✅ Gradle check result for a969638: SUCCESS

[codecov]

Codecov Report

❌ Patch coverage is 79.16667% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.40%. Comparing base (08ac04b) to head (a969638).

Files with missing lines	Patch %	Lines
.../index/engine/dataformat/merge/MergeScheduler.java	80.95%	4 Missing ⚠️
...opensearch/index/engine/DataFormatAwareEngine.java	66.66%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff            @@
##               main   #22319   +/-   ##
=========================================
  Coverage     73.39%   73.40%           
- Complexity    76048    76092   +44     
=========================================
  Files          6076     6076           
  Lines        345462   345485   +23     
  Branches      49725    49729    +4     
=========================================
+ Hits         253554   253589   +35     
+ Misses        71702    71646   -56     
- Partials      20206    20250   +44     

☔ View full report in Codecov by Harness.
📢 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.

[rayshrey]

Looks good to me, please attach the testing results

[Shailesh-Kumar-Singh]

Indexing throttling:

Ran clickbench ingestion twice.
First run was with max_merge_count - 1 (to simulate throttling) and second run with max_merge_count - 100 (to simulate no throttle). In both runs, the max thread count for merge was 1 (so that only 1 merge is active, and merges start queuing up).

In the first run we could see throttling while in the second run, there was no throttling and the number of segments were continuously building up.