Fix filter delegation concurrency bug by threading context_id through FFM upcalls by aravindsagar · Pull Request #21845 · opensearch-project/OpenSearch

aravindsagar · 2026-05-27T07:12:58Z

Description

FilterTreeCallbacks used global AtomicReference singletons for HANDLE and TRACKER. Under concurrent indexed-path queries, these were overwritten by the last query to enter startFragment, causing:

Query failures: collectDocs routed to wrong query's Lucene handle -> -1
Tracking mis-attribution: trackEnd routed to wrong task -> AssertionError

Fix: pass context_id (= OpenSearch task ID, already available in Rust from QueryTrackingContext) as the first parameter of every FFM upcall. Java uses it to look up the correct (handle, tracker) pair from a ConcurrentHashMap keyed by contextId. Each query gets isolated bindings.

Additionally guards trackStart against same-thread double-tracking (the original Slack-reported variant) by checking isThreadTrackedForTask before calling taskExecutionStartedOnThread.

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 · 2026-05-27T07:13:59Z

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit d2def46.

Path	Line	Severity	Description
server/src/main/java/org/opensearch/tasks/TaskResourceTrackingService.java	236	medium	Previously private method `isCurrentThreadWorkingOnTask` is refactored and re-exposed as the public method `isThreadTrackedForTask(Task task, long threadId)`. This widens the API surface of the task tracking subsystem, allowing any caller with a Task reference to probe whether an arbitrary thread ID is actively tracked — information that could be used to infer monitoring state or evade detection in adversarial code running within the same JVM.
sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsSearchService.java	223	low	A new hard `throw new IllegalStateException` is introduced when `task == null` during filter delegation setup. This silently changes the behavior for any prior code path where task could legitimately be null (previously it would proceed without tracking). If any production caller reaches this path with a null task, it becomes an unchecked runtime failure rather than a graceful no-op, creating a potential availability issue. The intent appears legitimate (enforcing invariants for per-query isolation), but the behavioral change is not backward-compatible.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 0 | Medium: 1 | Low: 1

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

github-actions · 2026-05-27T09:07:37Z

PR Reviewer Guide 🔍

(Review updated until commit `6347213`)

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 The code checks `if (task == null)` after already dereferencing `task.getId()` on line 324 (in the old code path). The new code moves the null check earlier (line 331), which is correct, but the old path at line 324 would have thrown NullPointerException if task was null when delegation is present. This suggests the null check may have been needed earlier in the original code as well, but the PR description does not mention this as a pre-existing bug being fixed. if (task == null) { throw new IllegalStateException("Filter delegation requires a tracked task for per-query isolation"); } Resource Leak Risk If `stream.close()` or `engine.close()` throw an exception, the `onClose` callback (which unregisters the per-query binding via `FilterTreeCallbacks.unregister`) may not run. The new close order executes `onClose` after closing stream and engine, so an exception in those close methods would skip the binding cleanup, leaking the entry in `FilterTreeCallbacks.BINDINGS`. The comment explains the ordering is needed for release upcalls, but does not address what happens if stream/engine close fails before reaching onClose. Exception first = closeQuietly(stream, null); first = closeQuietly(engine, first); first = closeQuietly(rowIdVector, first); if (onClose != null) { try { onClose.run(); } catch (Exception e) { if (first == null) first = e; else first.addSuppressed(e); } } Assertion Misuse The code uses assertions to detect lifecycle bugs (missing register, double register, premature unregister). Assertions are disabled by default in production Java (`-ea` is off), so these checks only run in tests. If a lifecycle bug occurs in production (e.g., a stale Rust handle outliving its query), the assertion is skipped, the binding lookup returns null, and the method silently returns -1. This hides the root cause in production while surfacing it in tests. The comment acknowledges this is intentional, but it means production deployments will see silent query failures (-1 from upcalls) rather than clear errors when bindings are missing. private static void assertBindingExists(QueryBinding binding, String op, long contextId) { assert binding != null : "FilterTreeCallbacks." + op + ": no binding for contextId=" + contextId + " (registered: " + BINDINGS.keySet() + ")"; }

github-actions · 2026-05-27T09:08:13Z

PR Code Suggestions ✨

Latest suggestions up to 6347213

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Ensure unregister completes even if close fails If `handle.close()` throws an exception, the cleanup lambda continues normally but the handle may be left in an inconsistent state. Consider whether the exception should be propagated or if additional cleanup is needed to prevent resource leaks when close fails. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java [837-852] public Runnable configureFilterDelegation( long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker, BackendExecutionContext backendContext ) { FilterTreeCallbacks.register(contextId, handle, tracker); return () -> { - FilterTreeCallbacks.unregister(contextId); try { - handle.close(); - } catch (Exception e) { - LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + FilterTreeCallbacks.unregister(contextId); + } finally { + try { + handle.close(); + } catch (Exception e) { + LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + } } }; } Suggestion importance[1-10]: 8 __ Why: This is a critical correctness issue. If `handle.close()` throws an exception, the cleanup continues but the binding remains registered in `FilterTreeCallbacks`, causing a resource leak. The improved code wraps `unregister` in a try-finally block to ensure it always executes, preventing leaked bindings that would cause assertion failures on subsequent queries with the same `contextId`.	Medium
	Ensure FFM upcalls complete before cleanup The close order assumes native release upcalls complete synchronously during stream/engine close, but if they're async or delayed, the binding could be unregistered before they execute. Consider adding a synchronization barrier or timeout after closing stream/engine to ensure all pending upcalls complete before running onClose. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/FragmentResources.java [79-95] public void close() throws Exception { - // Close the stream and engine first so any in-flight release upcalls from native - // code (e.g. ProviderHandle::drop -> releaseProvider) can still find their - // per-query binding in FilterTreeCallbacks. Running onClose first would unregister - // the binding while release upcalls are still pending, causing the eager release - // of Lucene resources to be skipped. Exception first = closeQuietly(stream, null); first = closeQuietly(engine, first); first = closeQuietly(rowIdVector, first); + // TODO: Add barrier here to ensure all pending FFM upcalls complete + // before unregistering the binding in onClose if (onClose != null) { try { onClose.run(); } catch (Exception e) { if (first == null) first = e; else first.addSuppressed(e); } } Suggestion importance[1-10]: 7 __ Why: The suggestion raises a valid concern about potential race conditions if native release upcalls are asynchronous. The comment in the code assumes synchronous completion, but if upcalls can be delayed, the binding could be unregistered prematurely. This is a legitimate correctness issue that could lead to resource leaks, though the current implementation may work if upcalls are indeed synchronous.	Medium
	Avoid expensive keySet() in hot-path assertion The assertion message construction concatenates `BINDINGS.keySet()` which can be expensive when many queries are registered. Since this is called on every hot-path upcall (createCollector, collectDocs), consider caching the keySet or removing it from the assertion message to avoid performance overhead in debug builds. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { assert binding != null : "FilterTreeCallbacks." + op + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + + contextId; } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies a performance concern in the assertion message construction. Calling `BINDINGS.keySet()` on every hot-path upcall (createCollector, collectDocs) can be expensive when many queries are registered. However, since assertions are disabled in production (`-ea` off), this only affects debug builds and tests. The impact is moderate rather than critical.	Low

Previous suggestions

Suggestions up to commit 6347213

Category	Suggestion	Impact
General	Close handle before unregistering binding The cleanup action closes the handle after unregistering the binding. If native code still holds a reference to the handle and attempts an upcall between `unregister` and `handle.close()`, the upcall will fail to find the binding. Consider closing the handle before unregistering, or document why the current order is safe. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java [837-851] public Runnable configureFilterDelegation( long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker, BackendExecutionContext backendContext ) { FilterTreeCallbacks.register(contextId, handle, tracker); return () -> { - FilterTreeCallbacks.unregister(contextId); try { handle.close(); } catch (Exception e) { LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + } finally { + FilterTreeCallbacks.unregister(contextId); } }; } Suggestion importance[1-10]: 7 __ Why: This suggestion identifies a potential race condition where native upcalls could occur between `unregister` and `handle.close()`. Closing the handle first (which should trigger native cleanup) before unregistering the binding is a safer ordering that ensures any final upcalls can still find their binding. The suggested change improves correctness.	Medium
	Add synchronization for native callbacks The close order assumes that closing `stream` and `engine` will trigger all pending native release upcalls before `onClose` unregisters the binding. However, if native code holds references that outlive the stream/engine close, release upcalls could still fire after `onClose` runs. Consider adding explicit synchronization or a grace period to ensure all native callbacks complete before unregistering. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/FragmentResources.java [80-95] public void close() throws Exception { - // Close the stream and engine first so any in-flight release upcalls from native - // code (e.g. ProviderHandle::drop -> releaseProvider) can still find their - // per-query binding in FilterTreeCallbacks. Running onClose first would unregister - // the binding while release upcalls are still pending, causing the eager release - // of Lucene resources to be skipped. Exception first = closeQuietly(stream, null); first = closeQuietly(engine, first); first = closeQuietly(rowIdVector, first); + // Ensure native release callbacks complete before unregistering binding if (onClose != null) { try { + // Consider adding explicit flush/wait for native callbacks here onClose.run(); } catch (Exception e) { if (first == null) first = e; else first.addSuppressed(e); } } + ... Suggestion importance[1-10]: 5 __ Why: The suggestion raises a valid concern about native callback timing, but the PR's close order (stream/engine before `onClose`) is intentionally designed to handle this. The suggestion to add "explicit flush/wait" is vague and doesn't provide a concrete implementation. The existing order is documented and appears correct for the use case.	Low
	Use String.format for assertion message The assertion message construction concatenates `BINDINGS.keySet()` which creates a snapshot of keys at assertion time. If the binding was removed between the null check and the assertion message construction, the snapshot won't reflect the race condition. Consider capturing the keySet before the null check or using a more defensive message format. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { - assert binding != null : "FilterTreeCallbacks." - + op - + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + assert binding != null : String.format( + "FilterTreeCallbacks.%s: no binding for contextId=%d (registered at check time: %s)", + op, + contextId, + BINDINGS.keySet() + ); } Suggestion importance[1-10]: 3 __ Why: The suggestion to use `String.format` is a minor style improvement that doesn't materially affect functionality. The current string concatenation works correctly and the snapshot timing concern is not a real issue since assertions are for development-time debugging, not production race detection.	Low

Suggestions up to commit ade06b9

Category	Suggestion	Impact
Possible issue	Prevent resource leak from double-register The assertion detects double-register but only when `-ea` is enabled. In production (no `-ea`), a double-register silently overwrites the previous binding, leaking the old handle's resources. Add a runtime check that throws `IllegalStateException` to prevent resource leaks in production. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [78-81] public static void register(long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker) { QueryBinding prev = BINDINGS.put(contextId, new QueryBinding(handle, tracker)); - assert prev == null : "FilterTreeCallbacks.register: binding already present for contextId=" + contextId; + if (prev != null) { + throw new IllegalStateException("FilterTreeCallbacks.register: binding already present for contextId=" + contextId); + } } Suggestion importance[1-10]: 9 __ Why: This is a critical bug. The assertion only catches double-register in tests (with `-ea`). In production, a double-register silently overwrites the previous binding, leaking the old `FilterDelegationHandle` which may hold native resources. The runtime check prevents this resource leak in all environments.	High
Possible issue	Guarantee cleanup action runs The comment explains that `onClose` must run after stream/engine close to avoid premature unregister. However, if `closeQuietly(stream)` or `closeQuietly(engine)` throws and `onClose` is never called, the binding remains registered indefinitely, leaking memory. Wrap the entire sequence in try-finally to guarantee `onClose` executes. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/FragmentResources.java [79-95] public void close() throws Exception { - // Close the stream and engine first so any in-flight release upcalls from native - // code (e.g. ProviderHandle::drop -> releaseProvider) can still find their - // per-query binding in FilterTreeCallbacks. Running onClose first would unregister - // the binding while release upcalls are still pending, causing the eager release - // of Lucene resources to be skipped. - Exception first = closeQuietly(stream, null); - first = closeQuietly(engine, first); - first = closeQuietly(rowIdVector, first); - if (onClose != null) { - try { - onClose.run(); - } catch (Exception e) { - if (first == null) first = e; - else first.addSuppressed(e); + Exception first = null; + try { + first = closeQuietly(stream, null); + first = closeQuietly(engine, first); + first = closeQuietly(rowIdVector, first); + } finally { + if (onClose != null) { + try { + onClose.run(); + } catch (Exception e) { + if (first == null) first = e; + else first.addSuppressed(e); + } } } ... } Suggestion importance[1-10]: 8 __ Why: This is a critical correctness issue. If `closeQuietly(stream)` or `closeQuietly(engine)` throws an exception that escapes (though `closeQuietly` should catch all exceptions), the `onClose` cleanup action (which unregisters the per-query binding) would never run, causing a memory leak. The try-finally pattern ensures cleanup always executes.	Medium
General	Optimize assertion message construction The assertion message concatenates `BINDINGS.keySet()` which can be expensive when many queries are registered. Consider logging the keySet only when the assertion fails, or limit the output to avoid performance impact during assertion evaluation in production-like test environments. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { - assert binding != null : "FilterTreeCallbacks." - + op - + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + if (binding == null) { + String msg = "FilterTreeCallbacks." + op + ": no binding for contextId=" + contextId + " (registered: " + BINDINGS.keySet() + ")"; + assert false : msg; + } } Suggestion importance[1-10]: 4 __ Why: The suggestion correctly identifies that `BINDINGS.keySet()` is evaluated eagerly during assertion message construction. However, the impact is minimal since assertions are only enabled in tests, and the keySet size is typically small (one per concurrent query). The improved code pattern is slightly better but the performance gain is negligible.	Low
General	Propagate handle close failures If `handle.close()` throws, the exception is logged but swallowed. The caller (in a finally block) expects cleanup to complete without throwing. However, if `handle` holds critical native resources, silently ignoring close failures may leak memory or file descriptors. Consider re-throwing as unchecked exception after logging. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java [833-848] public Runnable configureFilterDelegation( long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker, BackendExecutionContext backendContext ) { FilterTreeCallbacks.register(contextId, handle, tracker); return () -> { FilterTreeCallbacks.unregister(contextId); try { handle.close(); } catch (Exception e) { - LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + LOGGER.error(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + throw new RuntimeException("Failed to close FilterDelegationHandle for contextId=" + contextId, e); } }; } Suggestion importance[1-10]: 3 __ Why: The suggestion to propagate close failures is questionable. The cleanup action runs in a finally block (in `FragmentResources.close()`), and throwing from finally can mask the original exception. The current approach of logging and swallowing is appropriate for cleanup code. The severity should be `error` instead of `warn`, but re-throwing is not advisable.	Low

Suggestions up to commit ade06b9

Category	Suggestion	Impact
General	Ensure cleanup runs in finally The `onClose` cleanup (which unregisters the FFM binding) runs after closing stream/engine, but if `closeQuietly(rowIdVector, first)` throws an exception that gets suppressed, the `onClose` still executes. If `rowIdVector.close()` triggers native callbacks, those callbacks would fail to find their binding. Consider wrapping all cleanup in a try-finally to ensure `onClose` always runs. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/FragmentResources.java [79-95] public void close() throws Exception { - // Close the stream and engine first so any in-flight release upcalls from native - // code (e.g. ProviderHandle::drop -> releaseProvider) can still find their - // per-query binding in FilterTreeCallbacks. Running onClose first would unregister - // the binding while release upcalls are still pending, causing the eager release - // of Lucene resources to be skipped. - Exception first = closeQuietly(stream, null); - first = closeQuietly(engine, first); - first = closeQuietly(rowIdVector, first); - if (onClose != null) { - try { - onClose.run(); - } catch (Exception e) { - if (first == null) first = e; - else first.addSuppressed(e); + Exception first = null; + try { + first = closeQuietly(stream, null); + first = closeQuietly(engine, first); + first = closeQuietly(rowIdVector, first); + } finally { + if (onClose != null) { + try { + onClose.run(); + } catch (Exception e) { + if (first == null) first = e; + else first.addSuppressed(e); + } } } + if (first != null) throw first; +} Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that `onClose` should run in a finally block to guarantee cleanup even if earlier close operations fail. This improves robustness by ensuring the FFM binding is always unregistered.	Medium
	Propagate cleanup exceptions to caller If `handle.close()` throws an exception, the cleanup lambda logs it but doesn't propagate. The caller (FragmentResources.close) expects cleanup exceptions to be thrown so they can be properly tracked. Consider rethrowing as RuntimeException or documenting that cleanup exceptions are intentionally swallowed. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java [833-848] public Runnable configureFilterDelegation( long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker, BackendExecutionContext backendContext ) { FilterTreeCallbacks.register(contextId, handle, tracker); return () -> { - FilterTreeCallbacks.unregister(contextId); try { - handle.close(); - } catch (Exception e) { - LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + FilterTreeCallbacks.unregister(contextId); + } finally { + try { + handle.close(); + } catch (Exception e) { + LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + throw new RuntimeException("Failed to close FilterDelegationHandle for contextId=" + contextId, e); + } } }; } Suggestion importance[1-10]: 4 __ Why: The suggestion proposes propagating exceptions from `handle.close()`, but the current design intentionally logs and swallows them to prevent cleanup failures from disrupting query completion. The improved code also has a logic error (unregister in try, close in finally without proper exception chaining).	Low
	Capture keySet before assertion The assertion message construction concatenates `BINDINGS.keySet()` which creates a snapshot at assertion time. If the binding was removed between the null check and assertion message construction, the keySet won't reflect the state at failure time. Consider capturing the keySet before the assertion or using a more defensive approach. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { - assert binding != null : "FilterTreeCallbacks." - + op - + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + if (binding == null) { + String registeredKeys = BINDINGS.keySet().toString(); + throw new AssertionError("FilterTreeCallbacks." + + op + + ": no binding for contextId=" + + contextId + + " (registered: " + + registeredKeys + + ")"); + } } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a minor race condition in assertion message construction, but the impact is negligible since assertions are only enabled in tests. The existing code is acceptable for debugging purposes.	Low

Suggestions up to commit 726ca01

Category	Suggestion	Impact
Possible issue	Prevent binding overwrite before assertion The assertion occurs after `BINDINGS.put()` has already replaced any existing binding, causing a potential resource leak if a binding was present. Move the duplicate-check before the put operation to prevent overwriting an existing binding when assertions are enabled. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [78-81] public static void register(long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker) { - QueryBinding prev = BINDINGS.put(contextId, new QueryBinding(handle, tracker)); - assert prev == null : "FilterTreeCallbacks.register: binding already present for contextId=" + contextId; + assert !BINDINGS.containsKey(contextId) : "FilterTreeCallbacks.register: binding already present for contextId=" + contextId; + BINDINGS.put(contextId, new QueryBinding(handle, tracker)); } Suggestion importance[1-10]: 9 __ Why: This is a critical bug. The current code calls `BINDINGS.put()` first, which overwrites any existing binding, then asserts `prev == null`. If a binding existed, it's already been replaced and potentially leaked before the assertion fires. The suggested fix using `containsKey()` before `put()` prevents the overwrite when assertions are enabled, catching the lifecycle bug without causing a leak.	High
General	Optimize assertion message construction The assertion message concatenates `BINDINGS.keySet()` which can be expensive and may cause contention on the ConcurrentHashMap during concurrent access. Consider removing the keySet() call from the assertion message or only including it when the assertion actually fails (e.g., via a conditional check before the assert). sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { - assert binding != null : "FilterTreeCallbacks." - + op - + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + if (binding == null) { + String msg = "FilterTreeCallbacks." + op + ": no binding for contextId=" + contextId + " (registered: " + BINDINGS.keySet() + ")"; + throw new AssertionError(msg); + } } Suggestion importance[1-10]: 4 __ Why: The suggestion correctly identifies that calling `BINDINGS.keySet()` in the assertion message can be expensive during concurrent access. However, the impact is limited since assertions are typically disabled in production, and the suggested fix (throwing `AssertionError` explicitly) changes the behavior when assertions are off. The original code only evaluates the message when assertions are enabled and the condition fails.	Low

Suggestions up to commit 726ca01

Category	Suggestion	Impact
Possible issue	Use putIfAbsent for atomic registration The assertion after `put()` creates a race window where a duplicate contextId could overwrite an existing binding before the assertion fires. Use `putIfAbsent()` instead to atomically check and insert, preventing the race condition. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [78-81] public static void register(long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker) { - QueryBinding prev = BINDINGS.put(contextId, new QueryBinding(handle, tracker)); + QueryBinding prev = BINDINGS.putIfAbsent(contextId, new QueryBinding(handle, tracker)); assert prev == null : "FilterTreeCallbacks.register: binding already present for contextId=" + contextId; } Suggestion importance[1-10]: 8 __ Why: This is a valid concurrency issue. Using `putIfAbsent()` instead of `put()` eliminates the race window where a duplicate `contextId` could overwrite an existing binding before the assertion fires. This is a correctness improvement for the concurrent registration path.	Medium
General	Ensure unregister completes before close If `handle.close()` throws an exception, the cleanup lambda logs but continues. However, if the handle holds critical native resources, a failed close might leak them. Consider whether the exception should be propagated or if additional cleanup is needed. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java [832-848] public Runnable configureFilterDelegation( long contextId, FilterDelegationHandle handle, DelegationThreadTracker tracker, BackendExecutionContext backendContext ) { FilterTreeCallbacks.register(contextId, handle, tracker); return () -> { - FilterTreeCallbacks.unregister(contextId); try { - handle.close(); - } catch (Exception e) { - LOGGER.warn(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + FilterTreeCallbacks.unregister(contextId); + } finally { + try { + handle.close(); + } catch (Exception e) { + LOGGER.error(new ParameterizedMessage("FilterDelegationHandle.close() failed for contextId={}", contextId), e); + } } }; } Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies that `unregister` should be protected by a finally block to ensure it always executes, even if `handle.close()` throws. This prevents binding leaks. However, the original code already has the correct ordering (unregister before close), so the improvement is primarily about exception safety rather than a fundamental logic error.	Medium
	Avoid expensive keySet() in assertion The assertion message concatenates `BINDINGS.keySet()` which can be expensive in production if assertions are accidentally enabled. Consider using a lazy evaluation pattern or removing the keySet() from the assertion message to avoid potential performance impact during assertion evaluation. sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java [118-126] private static void assertBindingExists(QueryBinding binding, String op, long contextId) { assert binding != null : "FilterTreeCallbacks." + op + ": no binding for contextId=" - + contextId - + " (registered: " - + BINDINGS.keySet() - + ")"; + + contextId; } Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that `BINDINGS.keySet()` in the assertion message could be expensive if assertions are enabled. However, assertions are typically disabled in production, and the keySet() provides valuable debugging information during development. The impact is moderate since it only affects assertion-enabled builds.	Low
	Add synchronization for native callbacks The close ordering assumes native code will complete all upcalls before `stream.close()` returns, but async native operations might still be in-flight. Consider adding explicit synchronization or a barrier to ensure all native callbacks complete before unregistering the binding in `onClose`. sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/FragmentResources.java [79-95] public void close() throws Exception { - // Close the stream and engine first so any in-flight release upcalls from native - // code (e.g. ProviderHandle::drop -> releaseProvider) can still find their - // per-query binding in FilterTreeCallbacks. Running onClose first would unregister - // the binding while release upcalls are still pending, causing the eager release - // of Lucene resources to be skipped. Exception first = closeQuietly(stream, null); first = closeQuietly(engine, first); first = closeQuietly(rowIdVector, first); + // Ensure all native operations complete before unregistering + if (stream != null) { + // Add explicit flush/barrier if native stream supports it + } if (onClose != null) { try { onClose.run(); } catch (Exception e) { if (first == null) first = e; else first.addSuppressed(e); } } ... } Suggestion importance[1-10]: 3 __ Why: The suggestion raises a valid concern about async native operations, but the improved code doesn't provide a concrete solution (just a comment placeholder). The PR's close ordering is intentional and documented, and the suggestion doesn't demonstrate that the current approach is incorrect or provide a working alternative.	Low

github-actions · 2026-05-27T09:10:57Z

❌ Gradle check result for 65055f8: 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?

aravindsagar · 2026-05-27T10:04:14Z

gradle-check failing due to Maven throttling. For example,

> Could not GET 'https://repo.maven.apache.org/maven2/com/diffplug/durian/durian-core/1.2.0/durian-core-1.2.0.pom'. Received status code 429 from server: Too Many Requests

github-actions · 2026-05-27T10:49:57Z

Persistent review updated to latest commit e028959

github-actions · 2026-05-27T10:55:10Z

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

… FFM upcalls FilterTreeCallbacks used global AtomicReference singletons for HANDLE and TRACKER. Under concurrent indexed-path queries, these were overwritten by the last query to enter startFragment, causing: - Query failures: collectDocs routed to wrong query's Lucene handle -> -1 - Tracking mis-attribution: trackEnd routed to wrong task -> AssertionError Fix: pass context_id (= OpenSearch task ID, already available in Rust from QueryTrackingContext) as the first parameter of every FFM upcall. Java uses it to look up the correct (handle, tracker) pair from a ConcurrentHashMap keyed by contextId. Each query gets isolated bindings. Additionally guards trackStart against same-thread double-tracking (the original Slack-reported variant) by checking isThreadTrackedForTask before calling taskExecutionStartedOnThread. Signed-off-by: Aravind Sagar <sagarara@amazon.com> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

The isThreadTrackedForTask check guarded against the same-thread double-tracking scenario from the original Slack report. That scenario is structurally impossible on current main: cpu_executor.spawn(...).await in df_execute_with_context (commit 19b99d8) ensures all FFM upcalls fire on datafusion-cpu workers, never on the runTask thread that TaskAwareRunnable pre-tracked. Removing the guard so the assertion will fire if a future Rust change reintroduces the synchronous path — silent no-op would hide the bug. Also reverts the public isThreadTrackedForTask method on TaskResourceTrackingService that was added to support the guard. Signed-off-by: Aravind Sagar <sagarara@amazon.com> Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Three changes from code review: 1. Fix cleanup ordering in FragmentResources.close() The previous order ran onClose (which calls FilterTreeCallbacks.unregister) BEFORE closing the result stream. Closing the stream is what triggers Rust to drop ProviderHandle/FfmSegmentCollector, which fire releaseProvider/ releaseCollector upcalls. With unregister already done, those upcalls found no binding and skipped the eager Lucene Weight/Scorer release. Reorder: close stream → engine → reader first, then run onClose. Release upcalls now find their binding and call handle.releaseProvider/Collector as intended. 2. Move null-task check before handle creation in AnalyticsSearchService Previously the IllegalStateException for null task fired AFTER getFilterDelegationHandle had already been called, leaking the handle. Move the check to the top of the delegation block so we never allocate resources we won't track. 3. Close FilterDelegationHandle in cleanup DataFusionAnalyticsBackendPlugin.configureFilterDelegation now returns a cleanup that both unregisters from FilterTreeCallbacks and closes the handle. Previously nothing closed the handle eagerly — its internal ConcurrentHashMaps relied on GC. Also clarifies the testSharedContextIdCausesDataCorruption javadoc to explain that each thread re-registers its own handle at the shared contextId, simulating the old AtomicReference race. Signed-off-by: Aravind Sagar <sagarara@amazon.com> Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

The header comment said "Four callback slots" but lists five (and the code defines five). Pre-existing typo, easy to fix while in this area. Signed-off-by: Aravind Sagar <sagarara@amazon.com>

github-actions · 2026-05-29T05:41:22Z

✅ Gradle check result for 6badb56: SUCCESS

codecov · 2026-05-29T05:41:57Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 73.44%. Comparing base (62311ad) to head (6347213).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21845      +/-   ##
============================================
- Coverage     73.47%   73.44%   -0.04%     
+ Complexity    75576    75504      -72     
============================================
  Files          6034     6035       +1     
  Lines        342661   342710      +49     
  Branches      49294    49298       +4     
============================================
- Hits         251776   251698      -78     
- Misses        70901    70963      +62     
- Partials      19984    20049      +65

☔ 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.

himshikhagupta

Minor comment, mostly LGTM

Bukhtawar

Thanks for the fix. One suggestion: the current code silently returns -1 when a binding is missing, which makes contract violations invisible in testing. Adding assertions would catch lifecycle bugs (double-register, leaked bindings, premature unregister) during development while keeping production behavior safe.

Production keeps its silent fallbacks (return -1 / no-op) so a misuse never crashes the JVM through an FFM upcall. With assertions enabled (-ea, default in tests and ./gradlew run), the callbacks now also fail loudly on lifecycle violations so they surface in development: - register: asserts no prior binding exists for the contextId. Catches leaked bindings from earlier queries (missing unregister) and duplicate register calls. - createProvider/createCollector/collectDocs/release*: assert a binding exists for the contextId. Catches premature unregister and stale Rust handles outliving their query. The upcall methods catch Throwable but re-throw AssertionError so the assertion isn't swallowed by the surrounding error-handling block. In production (no -ea), these branches never execute. Replaces testNoHandleReturnsNegativeOne / testReleaseWithNoHandleIsSafe with testUnregisteredContextIdAsserts (now expects AssertionError). Replaces testSharedContextIdCausesDataCorruption (probabilistic race test) with testDoubleRegisterAsserts (deterministic — the assertion catches the bug at the API boundary). Signed-off-by: Aravind Sagar <sagarara@amazon.com> Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

aravindsagar · 2026-05-31T05:25:32Z

One suggestion: the current code silently returns -1 when a binding is missing, which makes contract violations invisible in testing. Adding assertions would catch lifecycle bugs (double-register, leaked bindings, premature unregister) during development while keeping production behavior safe.

Thanks Bukhtawar, added

github-actions · 2026-05-31T05:26:05Z

Persistent review updated to latest commit c05f752

github-actions · 2026-05-31T06:24:15Z

✅ Gradle check result for c05f752: SUCCESS

himshikhagupta

LGTM

aravindsagar · 2026-05-31T10:14:11Z

> Task :sandbox:qa:analytics-engine-coordinator:internalClusterTest

ShardFailoverIT > testQuerySucceedsAfterPrimaryNodeIsolated FAILED
    java.lang.NullPointerException: Cannot invoke "java.lang.Number.longValue()" because "java.util.List.get(int)[idx]" is null
        at __randomizedtesting.SeedInfo.seed([F42EACCE0A7FBDEB:8ED8E607B1353152]:0)
        at org.opensearch.analytics.resilience.ShardFailoverIT.testQuerySucceedsAfterPrimaryNodeIsolated(ShardFailoverIT.java:187)
REPRODUCE WITH: ./gradlew ':sandbox:qa:analytics-engine-coordinator:internalClusterTest' --tests 'org.opensearch.analytics.resilience.ShardFailoverIT.testQuerySucceedsAfterPrimaryNodeIsolated' -Dtests.seed=F42EACCE0A7FBDEB -Dtests.security.manager=true -Dtests.jvm.argline="-XX:TieredStopAtLevel=1 -XX:ReservedCodeCacheSize=64m" -Dtests.locale=en-VG -Dtests.timezone=America/Anchorage -Druntime.java=25

Not able to reproduce, the same test with the same seed succeeds locally. Seems like a flaky test

github-actions · 2026-05-31T10:20:57Z

Persistent review updated to latest commit c05f752

github-actions · 2026-06-01T05:30:59Z

Persistent review updated to latest commit 726ca01

github-actions · 2026-06-01T05:35:53Z

❌ Gradle check result for 726ca01: 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 · 2026-06-01T05:43:31Z

Persistent review updated to latest commit 726ca01

github-actions · 2026-06-01T06:32:27Z

❌ Gradle check result for 726ca01: 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?

aravindsagar · 2026-06-01T06:36:07Z

> Task :plugins:ingestion-kafka:internalClusterTest

Tests with failures:
 - org.opensearch.plugin.kafka.IngestPipelineFromKafkaIT.testFieldMappingDeleteWithPipeline

65 tests completed, 1 failed

Unrelated flaky test

github-actions · 2026-06-01T08:04:48Z

❕ Gradle check result for ade06b9: UNSTABLE

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

aravindsagar · 2026-06-01T08:47:23Z

Tests with failures:
 - org.opensearch.composite.CompositeParquetIndexIT.classMethod

FAILURE: Build failed with an exception.

* What went wrong:
Execution failed for task ':sandbox:plugins:composite-engine:internalClusterTest'.
> Test process encountered an unexpected problem.
   > class org.gradle.api.internal.tasks.testing.LifecycleTrackingTestEventReporter cannot be cast to class org.gradle.api.internal.tasks.testing.GroupTestEventReporterInternal

github-actions · 2026-06-01T08:49:18Z

Persistent review updated to latest commit ade06b9

github-actions · 2026-06-01T08:52:46Z

Persistent review updated to latest commit 6347213

github-actions · 2026-06-01T09:56:36Z

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

aravindsagar · 2026-06-01T09:58:27Z

> Task :example-plugins:rest-handler:javaRestTest

REPRODUCE WITH: ./gradlew ':example-plugins:rest-handler:javaRestTest' --tests 'org.opensearch.example.resthandler.ExampleFixtureIT.testExample' -Dtests.seed=EEDA83DEE087EF6A -Dtests.security.manager=true -Dtests.jvm.argline="-XX:TieredStopAtLevel=1 -XX:ReservedCodeCacheSize=64m" -Dtests.locale=ckb-IR -Dtests.timezone=Europe/Brussels -Druntime.java=25

ExampleFixtureIT > testExample FAILED
    java.lang.Exception: Test abandoned because suite timeout was reached.
        at __randomizedtesting.SeedInfo.seed([EEDA83DEE087EF6A]:0)

github-actions · 2026-06-01T10:52:35Z

✅ Gradle check result for 6347213: SUCCESS

… FFM upcalls (opensearch-project#21845) * Fix filter delegation concurrency bug by threading context_id through FFM upcalls FilterTreeCallbacks used global AtomicReference singletons for HANDLE and TRACKER. Under concurrent indexed-path queries, these were overwritten by the last query to enter startFragment, causing: - Query failures: collectDocs routed to wrong query's Lucene handle -> -1 - Tracking mis-attribution: trackEnd routed to wrong task -> AssertionError Fix: pass context_id (= OpenSearch task ID, already available in Rust from QueryTrackingContext) as the first parameter of every FFM upcall. Java uses it to look up the correct (handle, tracker) pair from a ConcurrentHashMap keyed by contextId. Each query gets isolated bindings. Additionally guards trackStart against same-thread double-tracking (the original Slack-reported variant) by checking isThreadTrackedForTask before calling taskExecutionStartedOnThread. Signed-off-by: Aravind Sagar <sagarara@amazon.com> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * Remove redundant double-tracking guard in DelegationThreadTracker The isThreadTrackedForTask check guarded against the same-thread double-tracking scenario from the original Slack report. That scenario is structurally impossible on current main: cpu_executor.spawn(...).await in df_execute_with_context (commit 19b99d8) ensures all FFM upcalls fire on datafusion-cpu workers, never on the runTask thread that TaskAwareRunnable pre-tracked. Signed-off-by: Aravind Sagar <sagarara@amazon.com>

aravindsagar requested a review from a team as a code owner May 27, 2026 07:12

gingeekrishna mentioned this pull request May 27, 2026

Remove legacy client benchmark and noop API plugin modules #21839

Merged

3 tasks

aravindsagar and others added 4 commits May 29, 2026 04:12

Fix stale module comment in ffm_callbacks.rs

6badb56

The header comment said "Four callback slots" but lists five (and the code defines five). Pre-existing typo, easy to fix while in this area. Signed-off-by: Aravind Sagar <sagarara@amazon.com>

aravindsagar force-pushed the fix/filter-delegation-concurrency branch from e028959 to 6badb56 Compare May 29, 2026 04:41

himshikhagupta reviewed May 30, 2026

View reviewed changes

Comment thread ...d-datafusion/src/main/java/org/opensearch/be/datafusion/indexfilter/FilterTreeCallbacks.java

himshikhagupta reviewed May 30, 2026

View reviewed changes

Bukhtawar reviewed May 30, 2026

View reviewed changes

aravindsagar and others added 2 commits May 31, 2026 05:09

Merge branch 'main' into fix/filter-delegation-concurrency

6f7b26d

himshikhagupta approved these changes May 31, 2026

View reviewed changes

Bukhtawar approved these changes May 31, 2026

View reviewed changes

aravindsagar closed this May 31, 2026

aravindsagar reopened this May 31, 2026

aravindsagar reopened this Jun 1, 2026

aravindsagar closed this Jun 1, 2026

aravindsagar reopened this Jun 1, 2026

Merge branch 'main' into fix/filter-delegation-concurrency

ade06b9

aravindsagar closed this Jun 1, 2026

aravindsagar reopened this Jun 1, 2026

Merge branch 'main' into fix/filter-delegation-concurrency

6347213

aravindsagar closed this Jun 1, 2026

aravindsagar reopened this Jun 1, 2026

Bukhtawar merged commit 989b7f6 into opensearch-project:main Jun 1, 2026
26 of 29 checks passed

sandeshkr419 mentioned this pull request Jun 2, 2026

Fix filter delegation collector reader mismatch #21934

Closed

3 tasks

Uh oh!

Conversation

aravindsagar commented May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Related Issues

Check List

Uh oh!

github-actions Bot commented May 27, 2026

PR Code Analyzer ❗

Uh oh!

github-actions Bot commented May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Reviewer Guide 🔍

(Review updated until commit 6347213)

Uh oh!

github-actions Bot commented May 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Code Suggestions ✨

Previous suggestions

Uh oh!

github-actions Bot commented May 27, 2026

Uh oh!

aravindsagar commented May 27, 2026

Uh oh!

github-actions Bot commented May 27, 2026

Uh oh!

github-actions Bot commented May 27, 2026

Uh oh!

github-actions Bot commented May 29, 2026

Uh oh!

codecov Bot commented May 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

himshikhagupta left a comment

Choose a reason for hiding this comment

Uh oh!

Bukhtawar left a comment

Choose a reason for hiding this comment

Uh oh!

aravindsagar commented May 31, 2026

Uh oh!

github-actions Bot commented May 31, 2026

Uh oh!

github-actions Bot commented May 31, 2026

Uh oh!

himshikhagupta left a comment

Choose a reason for hiding this comment

Uh oh!

aravindsagar commented May 31, 2026

Uh oh!

github-actions Bot commented May 31, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

aravindsagar commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

aravindsagar commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

aravindsagar commented Jun 1, 2026

Uh oh!

github-actions Bot commented Jun 1, 2026

Uh oh!

Uh oh!

aravindsagar commented May 27, 2026 •

edited

Loading

github-actions Bot commented May 27, 2026 •

edited

Loading

(Review updated until commit `6347213`)

github-actions Bot commented May 27, 2026 •

edited

Loading

codecov Bot commented May 29, 2026 •

edited

Loading