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 56d3574)

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 Possible Issue When activateThrottling throws an exception, isThrottling is reset to false, but when deactivateThrottling throws, isThrottling is reset to true. This asymmetry means a failed deactivation leaves throttling logically active (isThrottling=true) even though the actual throttle may not be engaged, potentially blocking future deactivation attempts until conditions change again. 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) { isThrottling.set(false); 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) { isThrottling.set(true); logger.warn("exception in deactivateThrottling callback", e); } } } } Race Condition evaluateThrottle is synchronized, but isThrottling is an AtomicBoolean accessed via getAndSet. If two threads call evaluateThrottle concurrently (e.g., from triggerMerges and the merge completion path), one thread could read stale isThrottling state before the other's callback completes, leading to duplicate or missed throttle transitions. 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) { isThrottling.set(false); 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) { isThrottling.set(true); logger.warn("exception in deactivateThrottling callback", e); } } } } Possible Issue In executeMerge, if submitMergeTask throws an exception, activeMerges is decremented but evaluateThrottle is not called. This means throttle state may remain active even after merge pressure drops below the threshold due to the exception, causing unnecessary indexing serialization until the next merge event. private void executeMerge() { while (activeMerges.get() < maxConcurrentMerges && mergeHandler.hasPendingMerges()) { OneMerge oneMerge = mergeHandler.getNextMerge(); if (oneMerge == null) { return; } try { submitMergeTask(oneMerge); } catch (Exception e) { activeMerges.decrementAndGet(); mergeHandler.onMergeFailure(oneMerge); onMergeFailureCleanup.run(); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 56d3574
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix incorrect counter decrement on exception The activeMerges counter is decremented in the exception handler, but it was never incremented before submitMergeTask was called. This causes the counter to become negative when an exception occurs during task submission. Move the increment inside submitMergeTask after successful task submission, or remove this decrement. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [323] } catch (Exception e) { - activeMerges.decrementAndGet(); mergeHandler.onMergeFailure(oneMerge); onMergeFailureCleanup.run(); } Suggestion importance[1-10]: 10 __ Why: Critical bug: activeMerges.decrementAndGet() is called when submitMergeTask throws an exception, but the counter was incremented earlier in executeMerge (line 315, in the while loop condition check). However, looking at the code flow, activeMerges is incremented in submitMergeTask itself (not shown in diff), so this decrement is actually correct to undo that increment when submission fails. But the suggestion correctly identifies that if the increment happens after successful submission, this decrement would cause the counter to go negative. This is a critical correctness issue that could lead to incorrect merge scheduling behavior.	High
General	Fix throttle deactivation threshold condition The throttle deactivation condition uses < instead of <=, creating a gap where throttling remains active when numMergesInFlight == maxMergeCount. This causes throttling to persist unnecessarily at the threshold. Change the condition to <= to deactivate throttling as soon as merge pressure drops to or below the limit. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [414] 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) { isThrottling.set(false); logger.warn("exception in activateThrottling callback", e); } } - } else if (numMergesInFlight < maxMergeCount) { + } else if (numMergesInFlight <= maxMergeCount) { Suggestion importance[1-10]: 7 __ Why: The condition numMergesInFlight < maxMergeCount creates a gap where throttling remains active when numMergesInFlight == maxMergeCount. Using <= would deactivate throttling at the threshold, which is more intuitive. However, the current behavior (keeping throttling active at the threshold) may be intentional to provide hysteresis and prevent rapid toggling.	Medium

---

Previous suggestions

Suggestions up to commit 0eee974

Category	Suggestion	Impact
Possible issue	Fix counter decrement without increment The activeMerges counter is decremented in the catch block after submitMergeTask fails, but it was never incremented before the task submission. This creates an inconsistency where the counter can become negative. Move the increment to occur before submitMergeTask is called. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [320-327] +activeMerges.incrementAndGet(); +try { + submitMergeTask(oneMerge); } catch (Exception e) { activeMerges.decrementAndGet(); mergeHandler.onMergeFailure(oneMerge); onMergeFailureCleanup.run(); } Suggestion importance[1-10]: 10 __ Why: Critical bug: activeMerges is decremented in the catch block without being incremented first, which can cause the counter to become negative. The increment happens inside submitMergeTask, so if that method throws before incrementing, the counter becomes inconsistent.	High
General	Fix throttle deactivation boundary condition The throttle deactivation condition uses strict inequality (<), creating a gap where throttling remains active when numMergesInFlight == maxMergeCount. This can cause unnecessary throttling at the threshold. Use <= to deactivate throttling when merges are at or below the limit. server/src/main/java/org/opensearch/index/engine/dataformat/merge/MergeScheduler.java [402-413] 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]: 7 __ Why: The condition numMergesInFlight < maxMergeCount creates a gap where throttling remains active when numMergesInFlight == maxMergeCount. Using <= would ensure throttling is deactivated at the threshold, though the current behavior may be intentional to provide hysteresis.	Medium

[github-actions]

❌ Gradle check result for 0eee974: 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 56d3574

[github-actions]

❌ Gradle check result for 56d3574: 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?