ESQL: Promptly clean up CCS exchange sinks by ebarlas · Pull Request #143325 · elastic/elasticsearch

ebarlas · 2026-02-28T03:38:25Z

Problem

The cross-cluster exchange sink created by ClusterComputeHandler for transferring reduced results from a remote cluster back to the origin coordinator was not explicitly finished on normal completion. It relied solely on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners.

Fix

Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern already used by DataNodeComputeHandler. Add an @After assertion to CrossClusterQueryIT to verify that all exchange sinks across all clusters are released within 5 seconds of each test.

This test pattern is already used in AbstractEsqlIntegTestCase.

@after

The cross-cluster exchange sink created by ClusterComputeHandler for transferring reduced results from a remote cluster back to the origin coordinator was not explicitly finished on normal completion. It relied solely on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern already used by DataNodeComputeHandler. Add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across all clusters are released within 5 seconds of each test.

elasticsearchmachine · 2026-02-28T03:40:52Z

Hi @ebarlas, I've created a changelog YAML for you.

idegtiarenko · 2026-03-02T13:14:18Z

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java

-                exchangeSink.addCompletionListener(computeListener.acquireAvoid());
+                exchangeSink.addCompletionListener(
+                    ActionListener.runBefore(computeListener.acquireAvoid(), () -> exchangeService.finishSinkHandler(globalSessionId, null))
+                );


This makes sense, but I would like @dnhatn to confirm

dnhatn

I've left a comment. Thanks for fixing this @ebarlas

dnhatn · 2026-03-02T20:23:35Z

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java

            );
            try (Releasable ignored = exchangeSource.addEmptySink()) {
-                exchangeSink.addCompletionListener(computeListener.acquireAvoid());
+                exchangeSink.addCompletionListener(


Can we revert this change and add this in 264 instead?

exchangeSink.addCompletionListener(ActionListener.running( () -> exchangeService.finishSinkHandler(globalSessionId, null));

elasticsearchmachine · 2026-03-02T22:07:01Z

Pinging @elastic/es-analytical-engine (Team:Analytics)

dnhatn

👍 Thanks @ebarlas!

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

elasticsearchmachine · 2026-03-03T01:49:36Z

💚 Backport successful

Status	Branch	Result
✅	9.3
✅	8.19
✅	9.2

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

…cations * upstream/main: (56 commits) Mute org.elasticsearch.compute.lucene.read.ValueSourceReaderTypeConversionTests testLoadAll elastic#143471 [DOCS] Fix ES|QL function and commands lists versioning metadata (elastic#143402) Fix MMROperatorTests (elastic#143453) Fix CSV-escaped quotes in generated docs examples (elastic#143449) Fix SQL client parsing of array header values (elastic#143408) ESQL: Add extended distribution tests and fault injection for external sources (elastic#143420) ESQL: Fix datasource test failures on Windows and FIPS (elastic#143417) Add circuit breaker for query construction to prevent OOM from automaton-based queries (elastic#142150) Cleanup SpecIT logging configuration (elastic#143365) ESQL: Prune unused regex extract nodes in optimizer (elastic#140982) Ensure supported locale outside of Entitlements check (elastic#143405) feat(es|ql): add dense_vector support in coalesce (elastic#142974) [Test] Unmute SnapshotStressTestsIT (elastic#143359) Mute org.elasticsearch.xpack.esql.CsvIT test {csv-spec:lookup-join.LookupJoinWithCoalesceFilterOnRight} elastic#143443 Mute org.elasticsearch.xpack.esql.CsvIT test {csv-spec:lookup-join.MvJoinKeyOnTheLookupIndex} elastic#143442 ESQL: Fix CCS exchange sink cleanup (elastic#143325) Mute org.elasticsearch.xpack.esql.CsvIT test {csv-spec:lookup-join.MvJoinKeyOnTheLookupIndexAfterStats} elastic#143434 Mute org.elasticsearch.xpack.esql.CsvIT test {csv-spec:lookup-join.MvJoinKeyFromRow} elastic#143432 Mute org.elasticsearch.xpack.esql.qa.mixed.MixedClusterEsqlSpecIT test {csv-spec:k8s-timeseries.Datenanos_derivative_compared_to_rate} elastic#143431 Mute org.elasticsearch.multiproject.test.CoreWithMultipleProjectsClientYamlTestSuiteIT test {yaml=search.retrievers/result-diversification/10_mmr_result_diversification_retriever/Test MMR result diversification single index float type} elastic#143430 ...

@after

The cross-cluster exchange sink in ClusterComputeHandler was not explicitly finished on normal completion. It relied on task cancellation or the InactiveSinksReaper for cleanup, unlike data node sinks which finish immediately via completion listeners. Add a finishSinkHandler call to the exchange sink's completion listener in runComputeOnRemoteCluster, matching the pattern used by DataNodeComputeHandler. Also add an @after assertion to CrossClusterQueryIT to verify that all exchange sinks across clusters are released within 5 seconds of each test.

ebarlas self-assigned this Feb 28, 2026

ebarlas added >bug v9.4.0 Team:Analytics Meta label for analytical engine team (ESQL/Aggs/Geo) :Analytics/ES|QL AKA ESQL labels Feb 28, 2026

Update docs/changelog/143325.yaml

7a5a8ef

elasticsearchmachine and others added 2 commits February 28, 2026 03:48

[CI] Auto commit changes from spotless

fb5e391

Merge branch 'main' into cleanup-cross-cluster-esql-exchange-sinks

b4c2ec9

idegtiarenko requested a review from dnhatn March 2, 2026 13:10

idegtiarenko reviewed Mar 2, 2026

View reviewed changes

dnhatn reviewed Mar 2, 2026

View reviewed changes

dnhatn self-requested a review March 2, 2026 20:23

Move addCompletionListener statement

2df75c9

ebarlas added auto-backport Automatically create backport pull requests when merged v9.3.2 v8.19.13 v9.2.7 labels Mar 2, 2026

ebarlas marked this pull request as ready for review March 2, 2026 22:06

dnhatn approved these changes Mar 2, 2026

View reviewed changes

ebarlas added 3 commits March 2, 2026 14:15

Merge branch 'main' into cleanup-cross-cluster-esql-exchange-sinks

59e46b3

Merge branch 'main' into cleanup-cross-cluster-esql-exchange-sinks

346e34a

Merge branch 'main' into cleanup-cross-cluster-esql-exchange-sinks

f3424ef

ebarlas merged commit 895a3b5 into elastic:main Mar 3, 2026
35 checks passed

ebarlas mentioned this pull request Mar 3, 2026

[9.3] ESQL: Fix CCS exchange sink cleanup (#143325) #143435

Merged

ebarlas mentioned this pull request Mar 3, 2026

[8.19] ESQL: Fix CCS exchange sink cleanup (#143325) #143436

Merged

ebarlas mentioned this pull request Mar 3, 2026

[9.2] ESQL: Fix CCS exchange sink cleanup (#143325) #143437

Merged

This was referenced Mar 6, 2026

[ML] Wait for cluster state in test #143767

Merged

[Transform] Disable PIT for CPS #143876

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ESQL: Promptly clean up CCS exchange sinks#143325

ESQL: Promptly clean up CCS exchange sinks#143325
ebarlas merged 8 commits intoelastic:mainfrom
ebarlas:cleanup-cross-cluster-esql-exchange-sinks

ebarlas commented Feb 28, 2026

Uh oh!

elasticsearchmachine commented Feb 28, 2026

Uh oh!

idegtiarenko Mar 2, 2026

Uh oh!

dnhatn left a comment

Uh oh!

dnhatn Mar 2, 2026

Uh oh!

elasticsearchmachine commented Mar 2, 2026

Uh oh!

dnhatn left a comment

Uh oh!

Uh oh!

elasticsearchmachine commented Mar 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

ebarlas commented Feb 28, 2026

Problem

Fix

Uh oh!

elasticsearchmachine commented Feb 28, 2026

Uh oh!

idegtiarenko Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

dnhatn left a comment

Choose a reason for hiding this comment

Uh oh!

dnhatn Mar 2, 2026

Choose a reason for hiding this comment

Uh oh!

elasticsearchmachine commented Mar 2, 2026

Uh oh!

dnhatn left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

elasticsearchmachine commented Mar 3, 2026

💚 Backport successful

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants