Module org.elasticsearch.compute

Package org.elasticsearch.compute.operator.exchange

Class BidirectionalBatchExchangeServer

java.lang.Object

org.elasticsearch.compute.operator.exchange.BidirectionalBatchExchangeServer

All Implemented Interfaces:

Closeable, AutoCloseable, org.elasticsearch.core.Releasable

public final class BidirectionalBatchExchangeServer
extends Object

Server-side handler for bidirectional batch exchange.

The server:

• Receives batches from the client via clientToServer exchange (using ExchangeSource)
• Sends results to the client via serverToClient exchange (using ExchangeSink)
• Uses BatchDriver to process batches
• Sends empty marker page when batch completes (via onBatchDone callback)

Only one batch can be active at a time. BatchDriver will throw an exception if multiple batches are sent concurrently.

Field Summary

Fields

Modifier and Type	Field	Description
protected final ExchangeService	exchangeService
protected final Executor	executor
protected final int	maxBufferSize
protected final String	sessionId
protected final Settings	settings
protected final Task	task
protected final TransportService	transportService

Constructor Summary

Constructors

Constructor	Description
BidirectionalBatchExchangeServer(String sessionId, String clientToServerId, String serverToClientId, ExchangeService exchangeService, Executor executor, int maxBufferSize, TransportService transportService, Task task, DiscoveryNode clientNode, Settings settings)	Create a new BidirectionalBatchExchangeServer with explicit exchange IDs.

Method Summary

Modifier and Type	Method	Description
protected static String	buildClientToServerId(String sessionId)	Constructs the client-to-server exchange ID from the session ID.
protected static String	buildServerToClientId(String sessionId)	Constructs the server-to-client exchange ID from the session ID.
void	close()
protected void	connectRemoteSink(DiscoveryNode node, String exchangeId, ExchangeSourceHandler sourceHandler, ActionListener<Void> listener, String errorMessagePrefix)	Connects a remote sink to a source handler via transport.
ExchangeSourceOperator.ExchangeSourceOperatorFactory	getSourceOperatorFactory()	Get the source operator factory for use in planning.
void	handleBatchExchangeStatusRequest(BatchExchangeStatusRequest request, TransportChannel channel, Task task)	Handle BatchExchangeStatusRequest from the client.
void	startWithOperators(DriverContext driverContext, ThreadContext threadContext, List<Operator> intermediateOperators, String clusterName, org.elasticsearch.core.Releasable releasable, ActionListener<Void> readyListener)	Start batch processing with the intermediate operators.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

sessionId

protected final String sessionId

exchangeService

protected final ExchangeService exchangeService

executor

protected final Executor executor

maxBufferSize

protected final int maxBufferSize

transportService

protected final TransportService transportService

task

protected final Task task

settings

protected final Settings settings

Constructor Details

BidirectionalBatchExchangeServer

public BidirectionalBatchExchangeServer(String sessionId,
 String clientToServerId,
 String serverToClientId,
 ExchangeService exchangeService,
 Executor executor,
 int maxBufferSize,
 TransportService transportService,
 Task task,
 DiscoveryNode clientNode,
 Settings settings)

Create a new BidirectionalBatchExchangeServer with explicit exchange IDs. Call startWithOperators(DriverContext, ThreadContext, List, String, Releasable, ActionListener) to complete setup and start processing after planning is complete

Method Details

getSourceOperatorFactory

public ExchangeSourceOperator.ExchangeSourceOperatorFactory getSourceOperatorFactory()

Get the source operator factory for use in planning. This can be called after construction to get the factory before calling startWithOperators.

Returns:

the source operator factory

startWithOperators

public void startWithOperators(DriverContext driverContext,
 ThreadContext threadContext,
 List<Operator> intermediateOperators,
 String clusterName,
 org.elasticsearch.core.Releasable releasable,
 ActionListener<Void> readyListener)

Start batch processing with the intermediate operators. This must be called after planning is complete.

handleBatchExchangeStatusRequest

public void handleBatchExchangeStatusRequest(BatchExchangeStatusRequest request,
 TransportChannel channel,
 Task task)

Handle BatchExchangeStatusRequest from the client. Called by ExchangeService's singleton handler which routes requests to the appropriate server.

The server stores the response channel BEFORE starting processing, ensuring it can always reply if an error occurs. Processing only starts after this request is received.

close

public void close()

buildClientToServerId

protected static String buildClientToServerId(String sessionId)

Constructs the client-to-server exchange ID from the session ID.

buildServerToClientId

protected static String buildServerToClientId(String sessionId)

Constructs the server-to-client exchange ID from the session ID.

connectRemoteSink

protected void connectRemoteSink(DiscoveryNode node,
 String exchangeId,
 ExchangeSourceHandler sourceHandler,
 ActionListener<Void> listener,
 String errorMessagePrefix)

Connects a remote sink to a source handler via transport. This is a common pattern used by both server and client to establish transport-based connections for bidirectional exchange.

Always uses failFast=true so the source handler aborts immediately on sink failure. The caller collects the real error via the listener and an EsqlRefCountingListener whose FailureCollector picks it over the generic TaskCancelledException thrown by the aborted source.