OpenSearch Documentation

Technical Documentation: opensearch-project/OpenSearch

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (README.md) Copy Section

• Welcome!
• Project Resources
• Code of Conduct
• Security
• License
• Copyright
• Trademark

Welcome! Copy Chapter

OpenSearch is an open-source, enterprise-grade search and observability suite that brings order to unstructured data at scale.

Project Resources Copy Chapter

• Project Website
• Downloads
• Documentation
• Need help? Try Forums or Slack
• Contributing to OpenSearch
• Maintainer Responsibilities
• Release Management
• Admin Responsibilities
• Testing
• Security

Code of Conduct Copy Chapter

The project's Code of Conduct outlines our expectations for all participants in our community, based on the OpenSearch Code of Conduct. Please contact [email protected] with any additional questions or comments.

Security Copy Chapter

If you discover a potential security issue in this project we ask that you notify OpenSearch Security directly via email to [email protected]. Please do not create a public GitHub issue.

License Copy Chapter

This project is licensed under the Apache v2.0 License.

Copyright Copy Chapter

Copyright OpenSearch Contributors. See NOTICE for details.

Trademark Copy Chapter

OpenSearch is a registered trademark of LF Projects, LLC.

OpenSearch includes certain Apache-licensed Elasticsearch code from Elasticsearch B.V. and other source code. Elasticsearch B.V. is not the source of that other source code. ELASTICSEARCH is a registered trademark of Elasticsearch B.V.

Plugins Arrow Flight Rpc Docs Architecture (`plugins/arrow-flight-rpc/docs/architecture.md`) Copy Section

Arrow Flight RPC Node-to-Node Architecture

/* Detailed source-code truncated for AI context efficiency. */

Plugins Arrow Flight Rpc Docs Backpressure (`plugins/arrow-flight-rpc/docs/backpressure.md`) Copy Section

Producer Back-Pressure

FlightServerChannel.sendResponseBatch honours gRPC's isReady() contract: the
producer thread parks on BackpressureStrategy.waitForListener before each batch
is queued, and resumes only once gRPC reports the per-stream outbound buffer has
drained below setOnReadyThreshold. Slow consumers throttle producer wall-clock
rather than allowing buffers to accumulate to the point of allocator OOM.

Why the eventloop queue exists Copy Chapter

Multiple producer threads may call channel.sendResponseBatch(...) concurrently
on the same stream (concurrent segment search, parallel batch generation). Arrow
Flight's ServerStreamListener.putNext is not safe for concurrent calls and
the contract requires start → putNext × N → completed in strict order on a
single thread; out-of-order or interleaved calls corrupt the stream.

Each FlightServerChannel therefore funnels submissions through a per-channel
single-threaded executor (the "eventloop"). Producer threads enqueue
BatchTasks; the eventloop dequeues them and performs the actual zero-copy
transfer plus putNext. This serialises ordering without making producer threads
contend on a per-channel mutex.

The back-pressure gate runs on the producer thread before the eventloop
submission, so a slow consumer throttles allocation rather than letting the
queue grow.

Settings Copy Chapter

• ready_timeout caps how long the producer thread parks before failing the
  batch with StreamErrorCode.TIMED_OUT. 100ms minimum.
• outbound_buffer_threshold is the per-stream gRPC buffered-bytes watermark.
  isReady() flips false once the per-stream outbound queue crosses this size.
  Must be strictly smaller than native.allocator.pool.flight.max so the
  gate engages before the allocator runs out (typical headroom: ~16 MiB).

Operator-facing behaviour Copy Chapter

Fast consumer (steady state) Copy Section

The readiness check is essentially free: the producer's call to
awaitReadyOrThrow returns immediately because gRPC's outbound buffer stays
below threshold.

Slow consumer Copy Section

• isReady() flips false once gRPC's per-stream outbound buffered-bytes crosses
  the threshold. The producer's next sendResponseBatch parks.
• gRPC fires OnReadyHandler after the consumer drains some bytes from the wire
  (HTTP/2 WINDOW_UPDATE arriving). The producer wakes and resumes.
• Sustained slowness for longer than ready_timeout causes the batch to fail
  with TIMED_OUT; the producer thread is freed and the stream terminates.

Cancellation Copy Section

Client-cancel propagates via gRPC's OnCancelHandler into the strategy's cancel
callback. Any thread parked on waitForListener wakes promptly with
StreamErrorCode.CANCELLED.

Tuning and operational concerns Copy Chapter

Sizing the flight pool Copy Section

native.allocator.pool.flight.max is per node, shared across all active
streams. outbound_buffer_threshold is per stream. Rule of thumb:

flight pool max  >=  N concurrent streams × (threshold + ~16 MiB headroom)

For N=10 concurrent streams at the default 64 MiB threshold, that's roughly
800 MiB. The threshold itself is a watermark, not a max-message-size — a single
batch larger than the threshold ships in one shot, with the producer parking on
the next sendResponseBatch. The hard constraint on per-batch size is the pool
cap (allocation must fit).

Thread-pool exhaustion under slow consumers Copy Section

The producer thread parks while waiting for the consumer. The action handler
runs on whichever thread pool it was registered against (typically SEARCH or
GENERIC); under N concurrent slow streams, N threads from that pool are
parked simultaneously. Once the pool is exhausted, new requests targeting it
queue up or are rejected.

Mitigations:

• Size ready_timeout to fail unresponsive streams within an acceptable window.
• Register stream actions on a pool sized for streaming workloads (not on
  shared CPU-bound pools): a pool of K threads both bounds concurrent streams
  to K and isolates the parking from unrelated traffic. A dedicated
  admission-control layer is a more flexible alternative, out of scope here.

⚠️ Known limitation: unbounded eventloop queue Copy Section

awaitReadyOrThrow checks listener.isReady(), which reflects only gRPC's
per-stream outbound buffer. The actual push to gRPC happens later, on the
per-channel eventloop, when the eventloop dequeues a BatchTask and calls
putNext. Between enqueue and putNext, the in-flight batch sits in our own
queue — invisible to gRPC, so isReady() doesn't account for it.

Implication: a producer that allocates batches significantly faster than the
eventloop drains can pile up retained batches in the queue before gRPC's
outbound crosses the threshold and isReady() flips false. The flight pool
allocator can be pushed past outbound_buffer_threshold by roughly the size
of the queued-but-not-yet-pushed batches. The current flight pool max sizing
guidance (threshold + ~16 MiB headroom per stream) absorbs typical bursts but
doesn't formally bound this.

A byte-aware bounded queue at the eventloop entry — that parks (or rejects)
the producer when the sum of queued batch sizes crosses a per-channel cap —
would close the gap for the async path. Open design questions:

• Cap dimension: byte-aware (sum of retained sizes) vs depth-aware (count).
  Bytes is correct for OOM protection.
• Mechanism: park the producer at enqueue (mirrors awaitReadyOrThrow's
  shape; same thread-pool-exhaustion caveat applies) vs reject with
  RESOURCE_EXHAUSTED.
• Sizing: per-channel cap (independent) vs carve from a shared per-node
  budget tied to flight pool max.

Tracked separately; not addressed in this change.

Synchronous send (bounded, opt-in) Copy Section

A caller that needs a hard bound today can send synchronously:
channel.sendResponseBatch(response, /* sync */ true). The batch is still
serialized and written on the channel's send executor (the same executor the
async path uses), but the calling thread blocks until the batch has been
pushed to gRPC's per-stream outbound buffer, so it cannot queue the next batch
ahead of this one — outstanding batches are bounded to one, with no eventloop
queue to grow. When that outbound buffer is full, the isReady() back-pressure
gate (above) throttles the producer.

A caller that opts in must drive a single stream from a single thread and must
not call from the send-executor thread itself. The default
sendResponseBatch(response) is unchanged — async, via the eventloop.

⚠️ Virtual threads for park-bound producers Copy Section

For workloads where the producer is mostly bottlenecked on
awaitReadyOrThrow (lots of parking, little compute per batch), the code
that registers the stream action can dispatch its handler on a virtual-thread
executor it owns. Park time then doesn't consume a platform thread.
CPU-bound work (e.g. building each batch) should still run on a sized
platform-thread pool to avoid pinning the carrier — typically by submitting
that work to a separate executor and awaiting the result from the virtual
thread. The framework itself does not assume virtual threads; this is a
decision for the action's registrant.

Plugins Arrow Flight Rpc Docs Channel Lifecycle Ownership (`plugins/arrow-flight-rpc/docs/channel-lifecycle-ownership.md`) Copy Section

FlightServerChannel Lifecycle & Buffer-Ownership Contract

Purpose Copy Chapter

Server-side Arrow Flight streaming frees off-heap Arrow buffers and terminates a gRPC
call from multiple unsynchronized threads at once. Without a single, explicit
ownership model this produces several classes of bug:

1. Double-close crash — close() runs its teardown twice, double-firing the
   TaskManager cancellable-channel untrack listener → AssertionError (with -ea) that
   kills the node.
2. Stream-root use-after-free / double-free — a concurrent cancel frees the reused
   stream root while the producer is mid-transferRoot or mid-putNext (gRPC still holds
   a zero-copy reference).
3. Stream-root / source-root leak — buffers transferred into the reused root after a
   concurrent close, or a source root for a batch whose send never runs, are owned by
   nobody.
4. Consumer hang — a batch send fails, the batch is dropped, no schema frame ever
   reaches the client, and the consumer's FlightStream.getRoot() blocks forever.

This document is the authoritative model. It was validated by an exhaustive
case-sweep (98 enumerated cases across buffer / thread-interleaving / method-exit /
lifecycle-state axes; 64 confirmed defects in the current code, collapsing to 12
distinct root-cause classes — see §9). The implementation in FlightServerChannel and
FlightOutboundHandler MUST conform to it, and §9's coverage matrix MUST stay true.

1. Threads Copy Chapter

All concurrency reduces to three thread classes per channel. The distinction between the
first two is load-bearing: the gap between them is where source roots can orphan (§6 S7).

Key consequences:

• The flight executor is single-threaded, so no two T_exec ops run concurrently with
  each other. Genuine concurrency is only T_caller/T_exec/bootstrap vs T_grpc.
• T_caller and T_exec are different threads: a batch can be parked in the gate on
  T_caller while a prior batch runs on T_exec. The window between "batch built on
  T_caller" and "sendBatch runs on T_exec" is where source roots orphan (§6 S7).

Single-writer model Copy Section

The stream root is the only state genuinely contended across threads: T_exec uses it
during transfer/putNext, while a cancel wants to free it. The channel resolves this by
confining the root to a single thread rather than synchronizing access to it:

• The stream root and every serverStreamListener call live only on T_exec — created,
  filled, and freed there. Because the executor is single-threaded, batch sends, terminal ops,
  and the root free are serialized by the executor itself; none can interleave.
• close() is a signal, not a buffer operation. From any thread it flips open once
  (CAS), fires close listeners, and posts the root free onto T_exec:
  executor.execute(() -> { if (root != null) { root.close(); root = null; } }). Because the
  executor is FIFO, that free runs after any in-flight send completes. close() touches no
  Arrow buffers and no gRPC from the caller's (possibly T_grpc) thread.

This is deadlock-free by construction: nothing is ever held across a gRPC/Arrow/blocking
call. The only mutex (closeListenerMutex, §4) is a leaf — its critical section only mutates an
ArrayList. close() from T_grpc never blocks: execute() enqueues and returns. There is no
waits-for cycle to construct. The single invariant a future change must preserve is the
self-evident "root and serverStreamListener are touched only on the flight executor."

The one trade-off is that close() frees the root asynchronously (on the executor) rather
than inline — see §10.

2. Roles & responsibilities Copy Chapter

Pivotal rule: the handler hands the channel a TransportResponse and a header
supplier, and receives back success or an exception. It never frees the stream root,
never serializes the header itself, and never closes the channel directly. Its one buffer
responsibility is the source-root hand-off (§3.1).

3. Buffer ownership — every buffer freed exactly once Copy Chapter

Zero-copy retention: native stream-root buffers are not freed right after putNext
— gRPC may still hold zero-copy refs. They are freed once, at close(), after the
framework's release boundary. The reused stream root lives from the first batch until
close().

3.1 Source-root hand-off discipline (closes S7 — the orphan class) Copy Section

The source root has two potential free sites on two threads, and the gap between
them is unguarded. Ownership transfers to the channel at sendResponseBatch entry, and
the handler guarantees the source is freed in exactly one of two mutually-exclusive
places via a handedOff flag:

sendResponseBatch(response, ...):                      // T_caller
  handedOff = false
  try:
    if (!(channel instanceof FlightServerChannel)) { relay error; return; }   // handedOff stays false
    flightChannel.awaitReadyOrThrow()                  // may throw CANCELLED/TIMED_OUT
    flightChannel.getExecutor().execute(task)          // may throw RejectedExecutionException
    handedOff = true                                   // sendBatch now owns it; its finally frees it
  finally:
    if (!handedOff) flightChannel.releaseUnsent(response)   // free here, exactly once

• Handed off → sendBatch runs on T_exec → its finally frees the source (§3.4).
• Not handed off (gate throws, execute rejects, or wrong-channel-type early return)
  → handler frees via releaseUnsent.
• Mutually exclusive ⇒ freed exactly once.

releaseUnsent(response) is a channel method (keeps "only the channel frees Arrow
buffers" intact): it closes response.getRoot() iff response instanceof ArrowBatchResponse (the byte path has no off-heap source). It does not touch the stream
root or lifecycle.

A second hand-off gap is inside processBatchTask on T_exec: getHeaderBuffer
(which throws IOException) and VectorSchemaRoot.create/transferRoot can throw
before the channel adopts the source. This is handled inside sendBatch itself
(§3.4), not by the handler — see the unified single-entry contract in §5.

3.2 Byte path: stream-root vector freed once, never per-batch Copy Section

The byte-serialized VectorStreamOutput.ByteSerialized vector is the stream root's
VarBinary vector once adopted. Freeing it per batch (today's try (out)) frees buffers
gRPC may still be draining (zero-copy) and races close(). The channel reuses the vector
across batches (reset()/setValueCount(0)) and frees it once in close().
ByteSerialized.close() must free only a vector it itself created (the first-batch,
existingRoot==null case); when adopting the channel's existing root it releases nothing.

3.3 Metadata buf: freed if `putNext` throws before hand-off Copy Section

buf = allocator.buffer(len); buf.writeBytes(metadata)
try { listener.putNext(buf) } catch (t) { if (buf.refCnt() > 0) buf.close(); throw t }

Covers the windows where Flight never adopts the buffer (waitUntilStreamReady throw,
ArrowMessage ctor throw). The refCnt() guard tolerates the narrow onNext-throw
subwindow where Flight's ArrowMessage.close() already released it.

3.4 Source-root free inside `sendBatch` (covers transfer/create/header throws) Copy Section

finally: if (native && sourceRoot != null) sourceRoot.close()

The native source root is owned exclusively by this sendBatch call, so it is freed
unconditionally in the finally on every exit: after a successful transfer it is empty
(close is a buffer no-op), and on any throw before the transfer — empty-field-vectors check,
VectorSchemaRoot.create OOM, transferRoot mid-loop, the header build — it still holds its
buffers (close frees them). A bare producerRoot.close() on the happy path plus a catch-only
cleanup is insufficient (today's bug): the free must be in a finally covering every exit.
On the native first batch, the freshly-created stream root is freed by
transferIntoStreamRoot's own try/finally if transfer fails before adoption, so an un-adopted
stream root cannot leak before becoming root.

3.5 Adopted stream root after a non-`Exception` `Throwable` in `sendBatch` Copy Section

serverStreamListener.start/putNext and transferRoot can throw a non-Exception
Throwable — OutOfMemoryError under memory pressure, or an AssertionError from Arrow/gRPC
under -ea. If this happens after the stream root is adopted (root != null),
sendBatch's finally frees only the source root; the stream root is freed only by
close(). The batch task is non-terminal, so BatchTask.close() does not release the channel —
nothing would call close() and the adopted stream root would leak. processBatchTask therefore
catches Throwable (after FlightRuntimeException/Exception) and routes it through failStream
(which releases the channel → close() posts the stream-root free) and reports it to the message
listener (wrapped to honour the listener's Exception contract) rather than letting it escape and
kill the executor worker. failStream closes the channel directly when there is no transport
channel, so close() runs on every variant.

private final AtomicBoolean open;          // "resources not yet freed"; flips true→false ONCE via CAS
private volatile boolean cancelled;        // set once by T_grpc in onChannelCancelled
private volatile VectorSchemaRoot root;    // reused native|byte stream root; T_exec-confined; nulled after free
private boolean terminalSent;              // completed() XOR error() issued; T_exec-confined
private final Object closeListenerMutex;   // leaf mutex: guards closeListeners + closeListenersFired only
private final List<ActionListener<Void>> closeListeners;
private boolean closeListenersFired;       // guarded by closeListenerMutex

• open — the teardown gate. compareAndSet(true,false) admits exactly one thread
  into teardown; every other close() caller returns immediately. The CAS winner is the sole
  poster of the root free and sole firer of close listeners. Thread-agnostic.
• cancelled — "client gone; gRPC already terminated the call." Read by sendBatch,
  completeStream, sendError to reject/no-op.
• root — touched only on T_exec (created, filled, freed there). volatile only so the
  test-only getRoot() reads a consistent reference; correctness rests on executor confinement,
  not volatility. Nulled immediately after free so nothing re-frees or re-fills it.
• terminalSent — T_exec-confined; at most one terminal gRPC op per stream.
• closeListenerMutex — a leaf mutex; its critical section only mutates the listener list +
  closeListenersFired, never a gRPC/Arrow/blocking call. It is the only mutex in the class.

Invariants (must hold under every interleaving) Copy Section

1. notifyCloseListeners() runs exactly once (CAS winner only) → TaskManager untrack
   fires once → no AssertionError, any thread, any order.
2. root.close() runs exactly once — only in the free task the CAS-winning close() posts
   to T_exec; root is nulled there. (If a send adopted no root, the task is a no-op.)
3. At most one terminal gRPC op (completed XOR error) per stream, and never after
   cancelled (completeStream/sendError check cancelled, both on T_exec).
4. No root freed under an in-flight send — the free runs on T_exec behind any in-flight
   send by the executor's FIFO ordering; root use and root free never overlap (single-writer).
5. No source root leaked or double-freed — freed exactly once via §3.1 (handoff) +
   §3.4 (in-send finally), mutually exclusive.
6. No deadlock — no lock is held across any gRPC/Arrow/blocking call; the only mutex
   (closeListenerMutex) is a leaf; close() only execute()s and returns. No waits-for cycle.
7. Close-listener registration is safe — addCloseListener either fires immediately
   (listeners already fired) or enqueues under closeListenerMutex, with no lost/duplicate fire.
8. A throwing close listener cannot skip the rest — notifyCloseListeners isolates each
   listener in try/catch so one failure can't strand TaskManager untrack.
9. Single-writer invariant (the property future changes must preserve): root and every
   serverStreamListener call are touched only on T_exec.

5. Method contracts Copy Chapter

All of sendBatch/completeStream/sendError run on T_exec and take no lock — the
single-threaded executor serializes them with each other and with the close()-posted root free.

`sendBatch(TransportResponse response, CheckedSupplier header)` — `T_exec`, no lock Copy Section

Single entry; native vs byte decided inside the channel. The header is built via a
supplier invoked before any root is mutated, so a header-build IOException fails fast —
the source is freed by the finally and no stream root is adopted. The handler never
serializes. The native source root is freed once on every exit by the outer finally
(unconditional: empty after a successful transfer → close is a no-op; full on an early throw →
close frees it).

sourceRoot = (response instanceof ArrowBatchResponse) ? response.getRoot() : null
try:
  if (cancelled)   throw StreamException.cancelled(...)   // reject → finally frees source
  if (!open.get()) throw IllegalStateException(...)       // reject → finally frees source
  firstBatch = (root == null)
  hdr = header.get()                                      // built FIRST; throw → finally, no root mutated
  if (native):
    if (firstBatch) root = create VSR from sourceRoot's allocator   // create+transfer guarded so an
    VectorTransfer.transferRoot(sourceRoot -> root)                 // un-adopted first root can't leak
  else (byte):
    if (firstBatch) root = new VarBinary VSR
    else            reset root's vector
    response.writeTo(serializer over root)
  if (firstBatch): middleware.setHeader(hdr); listener.start(root)
  if (metadata != null): buf=alloc; writeBytes; try{ listener.putNext(buf) } catch(t){ buf.close(); throw t }
  else: listener.putNext()
  batchNumber++; record stats
finally:
  if (native && sourceRoot != null) sourceRoot.close()    // every exit frees source ONCE

`completeStream(CheckedSupplier header)` — terminal, idempotent, `T_exec`, no lock Copy Section

if (!open.get() || cancelled || terminalSent) { record/log; return; }   // benign no-op
if (root == null) middleware.setHeader(header.get())   // empty stream
listener.completed(); terminalSent = true
recordCallEnd(OK)

Does not free root (free deferred to close(), §3 retention). The cancelled guard
closes the "terminal-after-cancel" race.

`sendError(CheckedSupplier header, Exception error)` — terminal, idempotent, `T_exec`, no lock Copy Section

if (!open.get() || cancelled || terminalSent) { record/log; return; }   // benign no-op
flightExc = map(error); middleware.setHeader(header.get())
listener.error(flightExc); terminalSent = true
recordCallEnd(mapped)

`releaseUnsent(TransportResponse response)` — `T_caller`/`T_exec` failed-handoff (§3.1) Copy Section

if (response instanceof ArrowBatchResponse r && r.getRoot() != null) r.getRoot().close();
// never touches stream root, terminalSent, or open

`close()` — fire-once, idempotent, ANY thread Copy Section

if (!open.compareAndSet(true, false)) return;          // losers exit at the door
executor.execute(() -> { if (root != null) { root.close(); root = null; } })   // free ON T_exec, behind any send
// (catch RejectedExecutionException on shutdown — see §10)
notifyCloseListeners();                                // runs exactly once; touches no Arrow buffers

close() posts the root free to T_exec rather than freeing inline, so it never touches Arrow
buffers or gRPC from the (possibly T_grpc) caller thread, and the free runs behind any
in-flight send by the executor's FIFO ordering.

`addCloseListener(listener)` — registration atomic vs `notifyCloseListeners` Copy Section

synchronized (closeListenerMutex):
  alreadyFired = closeListenersFired
  if (!alreadyFired) closeListeners.add(listener)
if (alreadyFired) listener.onResponse(null)            // fire outside the mutex

Prevents the register-vs-fire race (lost fire / double fire).

`notifyCloseListeners()` — fault-isolated Copy Section

synchronized (closeListenerMutex): snapshot = copy(closeListeners); clear(); closeListenersFired = true
for (l : snapshot) try { l.onResponse(null) } catch (e) { log; continue }   // one throw can't strand others

`onChannelCancelled()` — `T_grpc` Copy Section

if (cancelled) return;
cancelled = true; recordCallEnd(CANCELLED); close();

6. Scenario walk-through Copy Chapter

✓ = happens exactly once / as intended.

• S1 Normal multi-batch success — sendBatch×N (each frees its source ✓) →
  completeStream (completed() ✓) → release()→close() (free root ✓, notify ✓).
• S2 App error / batch-send failure — send throws → sendError (error() ✓) →
  release()→close(). Consumer gets an error frame, no hang.
• S3 Client cancel, idle channel — T_grpc close() frees root ✓, notify ✓. Later queued
  sendBatch sees cancelled → throws → finally frees source ✓. Later
  completeStream/sendError → cancelled no-op ✓.
• S4 Cancel races an in-flight sendBatch (UAF/double-free) — T_exec is mid
  transfer+putNext (occupying the single executor thread); T_grpc close() wins the CAS,
  posts the root free onto the busy executor, and returns without touching buffers. The
  free runs only after sendBatch finishes (FIFO) → frees root ✓ strictly after putNext. No
  UAF, freed once, notify once.
• S5 Cancel races a release()-driven close (the crash) — both call close();
  compareAndSet admits one; the other returns. notify once ✓ → no AssertionError.
• S6 Early/inbound error in getStream — listener.error then channel.close(); CAS
  admits one closer even if T_grpc also cancels ✓.
• S7 Cancel/timeout while a batch is parked in the gate (the orphan) —
  awaitReadyOrThrow() throws on T_caller before execute; handedOff==false →
  handler's finally calls releaseUnsent → source freed ✓. Same for execute rejection
  and the wrong-channel early return.
• S8 getHeaderBuffer/transferRoot/create throws inside sendBatch — lands in the
  finally; source freed once ✓; no stream root adopted (or freed by a later close() if
  adopted). No orphan.
• S9 Byte multi-batch + cancel — vector reused across batches on T_exec; freed once at
  close() (on T_exec, behind any send) ✓; no per-batch free, no write-vs-free race.
• S10 Non-Exception Throwable after stream-root adoption — putNext/start/transfer
  throws OOM/AssertionError once root != null. sendBatch's finally frees the source ✓;
  processBatchTask's catch (Throwable) runs failStream → channel released → close() posts
  the stream-root free ✓; listener notified, worker survives. No leak (§3.5).

7. Design decisions (explicit) Copy Chapter

1. Terminal op separate from root-free. completeStream/sendError issue the gRPC
   terminal op and set terminalSent; root is freed later in close(), preserving the
   zero-copy drain window. (Folding free into completeStream shrinks that window — not
   chosen.)
2. Single sendBatch(response, headerSupplier) entry. Native/byte decided in the
   channel; the handler never touches a stream root and never serializes the header.
3. Header built by the channel via a supplier, before any root is mutated. This is
   correctness-critical (not cosmetic): getHeaderBuffer throws IOException, and building it
   inside sendBatch puts that throw inside the source-freeing finally — and building it
   before the transfer means a header failure adopts no stream root.
4. Single-writer confinement, not synchronization. The stream root and all
   serverStreamListener calls are confined to the flight executor; close() posts the root
   free there rather than coordinating concurrent access to it. This makes the design
   deadlock-free by construction (nothing is held across a gRPC/Arrow call) at the cost of an
   asynchronous root free (§10). The invariant future changes must keep is the self-evident
   "root is touched only on the executor."

8. Relationship to the consumer-hang fix (PR #22117) Copy Chapter

PR #22117 fixes S2's symptom (a failed batch send calls sendError+release so the
consumer gets an error, not a hang) and relocates the native transfer into the channel.
That behaviour is preserved here. What this contract adds is the concurrency safety PR
#22117 lacks: it leaves close() non-atomic and the stream root accessible from both the
producer and the gRPC cancel thread, so S4 (UAF/double-free) and S5 (crash) remain reachable —
and its new failStream → releaseChannel → close() adds another concurrent close() caller.
This model closes S4/S5/S7–S9 while keeping S2 fixed.

9. Coverage matrix — the 12 distinct root-cause classes Copy Chapter

The case-sweep confirmed 64 defects in the current code; verified-and-deduplicated they are
12 classes. Each MUST map to a model mechanism. (# = confirmed-defect count folded in.)

10. Known limitations (not closed by this model) Copy Chapter

• Asynchronous stream-root free (a consequence of the single-writer choice). Because
  close() posts the root free onto the flight executor, the buffers are freed slightly after
  close() returns, not inline. Functionally this is the point (it serializes the free behind
  any in-flight send), but two implications follow: (a) tests asserting allocator state after
  close() must drain the executor first; (b) if the executor is already shut down when
  close() runs, the execute() is rejected and the stream root is not freed — see the next
  bullet. This is the deliberate trade the single-writer model makes for being deadlock-free by
  construction (§1, §7 decision 4).
• Node-shutdown drain gap. On executor.shutdownNow(), queued tasks — including a
  close()-posted root free, and any never-run BatchTasks — are dropped without running, so
  their roots are not freed. Likewise, if close()'s executor.execute(...) is rejected
  (executor already shut down), the root free never runs; close() logs this at WARN (not
  debug) so the shutdown-time leak is visible. We deliberately do not free the root inline on
  rejection: a rejection from shutdown() refuses new tasks while an already-running send may
  still be touching the root on the executor thread, so freeing from the (possibly gRPC) caller
  thread would reintroduce the use-after-free this model prevents. This is shutdown-only (the
  node is stopping; the OS reclaims off-heap memory on process exit), so impact is limited to
  noisy Arrow leak assertions / test flakiness, not a sustained production leak. Closing it fully
  would require draining queued tasks before shutdownNow; tracked separately.
• Final-frame zero-copy drain. Whether gRPC can still hold a reference to the last
  frame's buffers when close() frees root is a pre-existing concern with the same shape
  in today's code; out of scope here.

---

### Plugins Arrow Flight Rpc Docs Chaos (`plugins/arrow-flight-rpc/docs/chaos.md`)

# Chaos Testing

The Arrow Flight RPC plugin includes chaos testing capabilities to simulate network failures and test resilience.

## Enabling Chaos Testing

Chaos testing is disabled by default. To enable it, modify the `build.gradle` file:

### 1. Add Chaos Agent to internalClusterTest

Add this to the `internalClusterTest` task:

```gradle

internalClusterTest {
  // Enable chaos testing via bytecode injection
  doFirst {
    def agentJar = createChaosAgent()
    jvmArgs "-javaagent:${agentJar}"
  }
}

2. Add Chaos Agent Creation Task Copy Section

Add this task to create the chaos agent JAR:

// Task to create chaos agent JAR
def createChaosAgent() {
  def agentJar = file("${buildDir}/chaos-agent.jar")

  if (!agentJar.exists()) {
    def manifestFile = file("${buildDir}/MANIFEST.MF")
    manifestFile.text = '''Manifest-Version: 1.0
Premain-Class: org.opensearch.arrow.flight.chaos.ChaosAgent
Agent-Class: org.opensearch.arrow.flight.chaos.ChaosAgent
Can-Redefine-Classes: true
Can-Retransform-Classes: true
'''
    ant.jar(destfile: agentJar, manifest: manifestFile) {
      fileset(dir: sourceSets.internalClusterTest.output.classesDirs.first(), includes: 'org/opensearch/arrow/flight/chaos/ChaosAgent*.class')
    }
  }

  return agentJar.absolutePath
}

Running Chaos Tests Copy Chapter

Once enabled, run the chaos tests with:

./gradlew :plugins:arrow-flight-rpc:internalClusterTest --tests="*Chaos*"

What Chaos Testing Does Copy Chapter

The chaos testing framework:

• Injects bytecode to simulate network failures
• Tests client-side resilience to connection drops
• Validates proper error handling and recovery
• Ensures graceful degradation under adverse conditions

Plugins Arrow Flight Rpc Docs Error Handling (`plugins/arrow-flight-rpc/docs/error-handling.md`) Copy Section

Arrow Flight RPC Error Handling Guidelines

Overview Copy Chapter

This document describes the error handling model for the Arrow Flight RPC transport in OpenSearch. The model is inspired by gRPC's error handling approach and provides a consistent way to handle errors across the transport boundary.

At the OpenSearch layer, FlightRuntimeException isn't directly exposed. Instead, StreamException is used, which is converted to and from FlightRuntimeException at the flight transport layer.

Error Codes Copy Chapter

The following error codes are available in StreamErrorCode:

Best Practices Copy Chapter

Throwing Errors Copy Section

When throwing errors in server-side code:

// For validation errors
throw new StreamException(StreamErrorCode.INVALID_ARGUMENT, "Invalid parameter: " + paramName);

// For resource not found
throw new StreamException(StreamErrorCode.NOT_FOUND, "Resource not found: " + resourceId);

// For internal errors
throw new StreamException(StreamErrorCode.INTERNAL, "Internal error", exception);

// For unavailable resources
throw new StreamException(StreamErrorCode.UNAVAILABLE, "Resource temporarily unavailable");

// For cancelled operations
throw StreamException.cancelled("Operation cancelled by user");

Handling Errors Copy Section

When handling errors in client-side code:

try {
    // Operation that might throw StreamException
} catch (StreamException e) {
    switch (e.getErrorCode()) {
        case CANCELLED:
            // Handle cancellation
            break;
        case NOT_FOUND:
            // Handle resource not found
            break;
        case INVALID_ARGUMENT:
            // Handle validation error
            break;
        case UNAVAILABLE:
            // Handle temporary unavailability, maybe retry
            break;
        default:
            // Handle other errors
            break;
    }
}

Stream Cancellation Copy Section

When a stream is cancelled:

1. The client calls streamResponse.cancel(reason, cause)
2. The server receives a StreamException with StreamErrorCode.CANCELLED
3. The server should exit gracefully and not call completeStream() or sendResponse()

try {
    while (hasMoreData()) {
        channel.sendResponseBatch(createResponse());
    }
    channel.completeStream();
} catch (StreamException e) {
    if (e.getErrorCode() == StreamErrorCode.CANCELLED) {
        // Client cancelled - exit gracefully
        logger.info("Stream cancelled by client: {}", e.getMessage());
        // Do NOT call completeStream() or sendResponse()
        return;
    }
    // Handle other stream errors
    throw e;
}

Error Metadata Copy Chapter

StreamException supports adding metadata for additional error context:

StreamException exception = new StreamException(StreamErrorCode.INVALID_ARGUMENT, "Invalid query");
exception.addMetadata("query_id", queryId);
exception.addMetadata("index_name", indexName);
throw exception;

This metadata is preserved across the transport boundary and can be accessed on the receiving side:

Map<String, String> metadata = streamException.getMetadata();
String queryId = metadata.get("query_id");

Plugins Arrow Flight Rpc Docs Flight Client Channel Flow (`plugins/arrow-flight-rpc/docs/flight-client-channel-flow.md`) Copy Section

Flight Client Channel Stream Processing Flow and Error Handling

/* Detailed source-code truncated for AI context efficiency. */

Plugins Arrow Flight Rpc Docs Metrics (`plugins/arrow-flight-rpc/docs/metrics.md`) Copy Section

Arrow Flight RPC Metrics

The Arrow Flight RPC plugin provides comprehensive metrics to monitor the performance and health of the transport. These metrics are available through the Flight Stats API.

Accessing Metrics Copy Chapter

Metrics can be accessed using the Flight Stats API:

GET /_flight/stats

This returns metrics for all nodes. To get metrics for a specific node:

GET /_flight/stats/{node_id}

Monitoring Streaming Tasks Copy Chapter

Streaming transport tasks can be monitored using the existing Tasks API:

curl "localhost:9200/_cat/tasks?v"

Streaming tasks are identified by the stream-transport type:

action                                task_id                     parent_task_id              type             start_time    timestamp running_time ip        node
indices:data/read/search              TVk0SciMQtSwplV6rQwyMA:2165 -                           transport        1754082449785 21:07:29  169.5ms      127.0.0.1 node-1
indices:data/read/search[phase/query] TVk0SciMQtSwplV6rQwyMA:2166 TVk0SciMQtSwplV6rQwyMA:2165 stream-transport 1754082449786 21:07:29  168.4ms      127.0.0.1 node-1

Metrics Structure Copy Chapter

Metrics are organized into the following categories:

Client Call Metrics Copy Section

Metrics related to client-side calls:

Client Batch Metrics Copy Section

Metrics related to client-side batch operations:

Server Call Metrics Copy Section

Metrics related to server-side calls:

Server Batch Metrics Copy Section

Metrics related to server-side batch operations:

Status Metrics Copy Section

Metrics related to call status codes:

Resource Metrics Copy Section

Metrics related to resource usage:

Cluster-Level Metrics Copy Chapter

The API also provides cluster-level aggregated metrics that combine data from all nodes:

GET /_flight/stats

The response includes a cluster_stats section with aggregated metrics for:

• Client calls and batches (aggregated across all nodes)
• Server calls and batches (aggregated across all nodes)
• Average durations and throughput

Note: All duration and size fields include both human-readable formats (e.g., "1s", "24.6kb") and raw values in nanoseconds/bytes.

Example Response Copy Chapter

/* Detailed source-code truncated for AI context efficiency. */

Interpreting Metrics Copy Chapter

Performance Monitoring Copy Section

• High latency: Check duration metrics for client and server calls
• Memory pressure: Monitor arrow_allocated_bytes and arrow_peak_bytes
• Thread pool saturation: Check client_thread_utilization_percent and server_thread_utilization_percent

Error Detection Copy Section

• Failed calls: Monitor non-OK status counts in status.client and status.server
• Cancelled operations: Check CANCELLED status counts
• Resource exhaustion: Watch for RESOURCE_EXHAUSTED status counts

Throughput Analysis Copy Section

• Request throughput: Monitor client_calls.started and server_calls.started rates
• Data throughput: Track client_calls.request_bytes and server_batches.sent_bytes rates
• Batch efficiency: Compare client_batches.received with client_batches.requested

Plugins Arrow Flight Rpc Docs Native Arrow Transport Design (`plugins/arrow-flight-rpc/docs/native-arrow-transport-design.md`) Copy Section

Native Arrow Transport Path

Overview Copy Chapter

The Arrow Flight transport supports a native Arrow path where typed VectorSchemaRoot data
flows directly over Flight without byte serialization. This is for APIs that produce
Arrow-columnar data natively (e.g., query engines like DataFusion).

The existing byte-serialized path (writeTo/read via StreamOutput/StreamInput) is unchanged.

Quick Start Copy Chapter

1. Define your response Copy Section

Extend ArrowBatchResponse. No writeTo/read override needed — the framework handles it.

public class MyQueryResponse extends ArrowBatchResponse {
    public MyQueryResponse(VectorSchemaRoot root) { super(root); }
    public MyQueryResponse(StreamInput in) throws IOException { super(in); }
}

2. Server-side handler — produce Arrow data Copy Section

void handleRequest(MyRequest request, TransportChannel channel, Task task) throws IOException {
    // Get the channel's allocator. Use this directly for producer roots
    // (not a child allocator) to avoid Arrow's cross-allocator transfer bug
    // with foreign-backed buffers from C data import.
    BufferAllocator allocator = ArrowFlightChannel.from(channel).getAllocator();

    Schema schema = new Schema(List.of(
        new Field("name", FieldType.nullable(new ArrowType.Utf8()), null),
        new Field("score", FieldType.nullable(new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), null)
    ));

    try {
        for (int i = 0; i < batchCount; i++) {
            VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
            // populate vectors...
            channel.sendResponseBatch(new MyQueryResponse(root));
            // root is now owned by the framework — don't reuse or close it
        }
        // Cleanup callback runs on executor after all batches are flushed.
        channel.completeStream();
    } catch (Exception e) {
        channel.sendResponse(e);
    }
}

3. Client-side handler — consume Arrow data Copy Section

class MyQueryHandler implements StreamTransportResponseHandler<MyQueryResponse> {

    public MyQueryResponse read(StreamInput in) throws IOException {
        return new MyQueryResponse(in);
    }

    public void handleStreamResponse(StreamTransportResponse<MyQueryResponse> stream) {
        MyQueryResponse response;
        while ((response = stream.nextResponse()) != null) {
            VectorSchemaRoot root = response.getRoot();
            VarCharVector names = (VarCharVector) root.getVector("name");
            Float8Vector scores = (Float8Vector) root.getVector("score");
            // process typed vectors...
        }
        stream.close();
    }

    public void handleException(TransportException exp) { /* handle error */ }
    public String executor() { return ThreadPool.Names.GENERIC; }
}

Allocator Management Copy Chapter

The allocator used for producer roots must be long-lived — it must outlive the gRPC
stream. This is because gRPC's zero-copy write path (ArrowBufRetainingCompositeByteBuf)
retains ArrowBuf references beyond putNext() and completed(), releasing them
asynchronously on the Netty event loop. Closing the allocator while gRPC still holds
these references causes memory accounting errors.

Use ArrowFlightChannel.from(channel).getAllocator() to get the channel's allocator,
or use your own long-lived application allocator. Do not create and close a child
allocator per request.

Important: C Data Import and allocator choice Copy Section

If your producer imports data via Arrow's C Data Interface (Data.importIntoVector,
Data.importIntoVectorSchemaRoot), the imported buffers are foreign-backed. Arrow Java
has a bug where cross-allocator transferOwnership of foreign-backed buffers doesn't
properly release the internal ArrowArray C struct buffer (128 bytes per import call),
causing a memory leak in the source allocator.

The framework creates the shared Flight root from the producer's allocator (the
allocator of the first batch's vectors), ensuring same-allocator transfer which avoids
this bug. All subsequent batches should use the same allocator.

Ownership Contract Copy Chapter

After calling sendResponseBatch(response), the framework owns the response's root.
Do not reuse or close it — the framework transfers its buffers and closes it on the executor.

Pipelining Copy Chapter

Batches can be produced in parallel. Each batch must have its own VectorSchemaRoot
(created from the channel's allocator). The framework serializes the transfer and send
on the executor thread. The producer can queue batches without waiting for each to flush.

Application Metadata Copy Chapter

Each ArrowBatchResponse can carry an opaque byte[] of application metadata that
travels on the same Flight frame as its data — Arrow Flight's
OutboundStreamListener.putNext(ArrowBuf metadata) is what's used on the wire. The
transport does not interpret these bytes; encoding and decoding are the action's
responsibility.

Typical uses:

• Per-batch metadata keyed to this batch's data — row offsets, watermarks,
  batch-level stats.
• Stream-terminal payloads — attach to the last batch only. Profiling counters,
  cumulative stats, summary blobs.

// send-side: attach to the last batch
boolean isLast = (b == batchCount - 1);
if (isLast && profile) {
    byte[] profileBytes = serializeProfile(...);
    channel.sendResponseBatch(new MyQueryResponse(root, profileBytes));
} else {
    channel.sendResponseBatch(new MyQueryResponse(root));
}

// receive-side: read on every batch, act when present
while ((r = stream.nextResponse()) != null) {
    consume(r.getRoot());
    byte[] meta = r.getMetadata();
    if (meta != null) handleProfile(meta);
}

Lifecycle: bytes are copied off the Flight wire buffer into a byte[] owned by the
response, so the metadata outlives the stream cursor.

The metadata channel is exposed on ArrowBatchResponse and is only available on the
native Arrow path. The byte-serialized path does not surface it.

Plugins Arrow Flight Rpc Docs Netty4 Vs Flight Comparison (`plugins/arrow-flight-rpc/docs/netty4-vs-flight-comparison.md`) Copy Section

Netty4 vs Flight Transport Comparison

This document compares the traditional Netty4 transport with the new Arrow Flight transport across all four communication flows.

1. Outbound Client: Netty4 vs. Flight Copy Chapter

sequenceDiagram
    participant Client
    participant TS as TransportService
    participant CM as ConnectionManager
    participant C as Connection
    participant TC as TcpChannel(Netty4TcpChannel)
    participant NOH as NativeOutboundHandler
    participant N as Network

    Note over Client,N: Netty4 Flow
    Client->>TS: Send TransportRequest
    TS->>TS: Generate reqID
    TS->>CM: Get Connection
    CM->>C: Provide Connection
    C->>TC: Use Channel
    TC->>NOH: Serialize to BytesReference(StreamOutput) with reqID
    NOH->>N: Send BytesReference

    participant Client2
    participant STS as StreamTransportService
    participant CM2 as ConnectionManager
    participant C2 as Connection
    participant FTC as FlightTcpChannel
    participant FMH as FlightMessageHandler
    participant FC as FlightClientChannel
    participant N2 as Network

    Note over Client2,N2: Flight Flow
    Client2->>STS: Send TransportRequest
    STS->>STS: Generate reqID
    STS->>CM2: Get Connection
    CM2->>C2: Provide Connection
    C2->>FTC: Use Channel
    FTC->>FMH: Serialize to Flight Ticket(ArrowStreamOutput) with reqID
    FMH->>FC: Send Flight Ticket
    FC->>N2: Transmit Request

2. Inbound Server: Netty4 vs. Flight Copy Chapter

sequenceDiagram
    participant STC as Server TcpChannel(Netty4TcpChannel)
    participant IP as InboundPipeline
    participant IH as InboundHandler
    participant NMH as NativeMessageHandler
    participant RH as RequestHandler

    Note over STC,RH: Netty4 Flow
    STC->>IP: Receive BytesReference
    IP->>IH: Deserialize to InboundMessage(StreamInput)
    IH->>NMH: Interpret as TransportRequest
    NMH->>RH: Process Request

    participant FS as FlightServer
    participant FP as FlightProducer
    participant IP2 as InboundPipeline
    participant IH2 as InboundHandler
    participant NMH2 as NativeMessageHandler
    participant RH2 as RequestHandler

    Note over FS,RH2: Flight Flow
    FS->>FP: Receive Flight Ticket
    FP->>FP: Create VectorSchemaRoot
    FP->>FP: Create FlightServerChannel
    FP->>IP2: Pass to InboundPipeline
    IP2->>IH2: Deserialize with ArrowStreamInput
    IH2->>NMH2: Interpret as TransportRequest
    NMH2->>RH2: Process Request

3. Outbound Server: Netty4 vs. Flight Copy Chapter

sequenceDiagram
    participant RH as RequestHandler
    participant OH as OutboundHandler
    participant TTC as TcpTransportChannel
    participant TC as TcpChannel

    Note over RH,TC: Netty4 Flow
    RH->>TTC: sendResponse(TransportResponse)
    TTC->>OH: Serialize TransportResponse(via sendResponse)
    OH->>TC: Send Serialized Data to Client

    participant RH2 as RequestHandler
    participant FTC as FlightTransportChannel
    participant FOH as FlightOutboundHandler
    participant FSC as FlightServerChannel
    participant SSL as ServerStreamListener

    Note over RH2,SSL: Flight Flow
    RH2->>FTC: sendResponseBatch(TransportResponse)
    FTC->>FOH: sendResponseBatch
    FOH->>FSC: sendBatch(VectorSchemaRoot)
    FSC->>SSL: start(root) (first batch)
    FSC->>SSL: putNext() (stream batch)
    RH2->>FTC: completeStream()
    FTC->>FOH: completeStream
    FOH->>FSC: completeStream
    FSC->>SSL: completed() (end stream)

4. Inbound Client: Netty4 vs. Flight Copy Chapter

sequenceDiagram
    participant CTC as Client TcpChannel(Netty4TcpChannel)
    participant CIP as Client InboundPipeline
    participant CIH as Client InboundHandler
    participant RH as ResponseHandler

    Note over CTC,RH: Netty4 Flow
    CTC->>CIP: Receive BytesReference
    CIP->>CIH: Deserialize to TransportResponse(StreamInput)
    CIH->>RH: Deliver Response

    participant FC as FlightClient
    participant FCC as FlightClientChannel
    participant FTR as FlightTransportResponse
    participant RH2 as ResponseHandler

    Note over FC,RH2: Flight Flow (Async Response Handling)
    FC->>FCC: handleInboundStream(Ticket, Listener)
    FCC->>FTR: Create FlightTransportResponse
    FCC->>FCC: Retrieve Header and reqID
    FCC->>RH2: Get TransportResponseHandlerusing reqID
    FCC->>RH2: handler.handleStreamResponse(streamResponse)(Async Processing)

Key Differences Summary Copy Chapter

**Netty4 Transport (Traditional)**: Copy Section

• Request/Response: Single request → single response pattern
• Serialization: BytesReference with StreamOutput/StreamInput
• Channel: Netty4TcpChannel with native handlers
• Processing: Synchronous response handling
• Protocol: Custom binary protocol over TCP

**Flight Transport (New)**: Copy Section

• Streaming: Single request → multiple response batches
• Serialization: Arrow Flight Ticket with ArrowStreamOutput/ArrowStreamInput
• Channel: FlightClientChannel/FlightServerChannel with Flight handlers
• Processing: Asynchronous stream processing with nextResponse() loop
• Protocol: Arrow Flight RPC over gRPC

Plugins Arrow Flight Rpc Docs Security Integration (`plugins/arrow-flight-rpc/docs/security-integration.md`) Copy Section

Security Plugin Integration

The Arrow Flight RPC plugin integrates with the OpenSearch Security plugin to provide secure streaming transport with TLS encryption. The Flight transport operates on a dedicated port (default 9400-9500) for internal node-to-node communication, similar to the standard transport on port 9300.

⚠️ WARNING: Always use FLIGHT-SECURE with flight.ssl.enable: true when the security plugin is installed. Using FLIGHT (non-secure) mode bypasses transport-layer encryption and allows unencrypted node-to-node communication. See Transport Modes for details.

Configuration Copy Chapter

Add these settings to opensearch.yml:

# Enable streaming transport
opensearch.experimental.feature.transport.stream.enabled: true

# Use secure Flight as default transport
transport.stream.type.default: FLIGHT-SECURE

# Enable Flight TLS
flight.ssl.enable: true

Transport Modes: FLIGHT vs FLIGHT-SECURE Copy Chapter

The plugin provides two transport modes. Choosing the correct mode is critical for security:

FLIGHT-SECURE (Recommended) Copy Section

FLIGHT-SECURE enables TLS encryption on the Flight transport port, providing the same security guarantees as the standard transport port (9300) with plugins.security.ssl.transport.enabled: true. This is the only mode that should be used when the OpenSearch Security plugin is installed.

FLIGHT (Non-Secure) Copy Section

⚠️ WARNING: FLIGHT mode transmits all inter-node traffic in plaintext. Do not use this mode when the security plugin is installed or when the cluster handles sensitive data.

FLIGHT mode starts the Flight transport without TLS. When used with the security plugin installed, this creates a misconfiguration where:

• Node-to-node communication on the Flight port is unencrypted
• Any client that can reach the Flight port can send requests without TLS-based node identity verification

This mode exists only for development and testing environments where the security plugin is not installed.

Security Plugin Setup Copy Chapter

Install and configure the security plugin:

# Install security plugin
bin/opensearch-plugin install opensearch-security

# Setup demo configuration
plugins/opensearch-security/tools/install_demo_configuration.sh

Secure Configuration Checklist Copy Chapter

When deploying with the security plugin, verify all three settings are present:

# 1. Enable the experimental streaming transport feature
opensearch.experimental.feature.transport.stream.enabled: true

# 2. Select the secure transport mode (NOT "FLIGHT")
transport.stream.type.default: FLIGHT-SECURE

# 3. Enable TLS on the Flight port
flight.ssl.enable: true

Missing any of these settings results in an insecure configuration. Specifically:

• Omitting flight.ssl.enable: true — TLS defaults to disabled (false), leaving the Flight port unencrypted
• Using FLIGHT instead of FLIGHT-SECURE — starts the transport without TLS regardless of other settings

Role-Based Access Control Copy Chapter

The Flight transport supports all security plugin features:

• Index-level permissions
• Document-level security (DLS)
• Field-level security (FLS)
• Action-level permissions

These security controls are enforced at the action filter layer (SecurityFilter) and apply regardless of the transport type. RBAC, DLS, and FLS protections work correctly when the cluster is properly configured with FLIGHT-SECURE.

How Security Is Enforced Copy Chapter

The Flight transport reuses the same security enforcement as the standard Netty transport:

1. Connection layer (TLS/mTLS): When FLIGHT-SECURE is used with flight.ssl.enable: true, mutual TLS authenticates nodes at the connection level — the same model as port 9300 with plugins.security.ssl.transport.enabled: true.
2. Action layer (SecurityFilter): RBAC, DLS, and FLS are enforced by the security plugin's action filter, which runs independently of the transport channel type.

Both layers are required for a fully secure deployment. TLS ensures only trusted nodes can connect; the action filter ensures authorized access to data.

Plugins Arrow Flight Rpc Docs Server Side Streaming Guide (`plugins/arrow-flight-rpc/docs/server-side-streaming-guide.md`) Copy Section

Server-Side Streaming API Guide

Overview Copy Chapter

Server-side streaming allows sending multiple response batches to a client over a single connection. This is ideal for large result sets, real-time data, or progressive processing.

Action Registration Copy Chapter

streamTransportService.registerRequestHandler(
    "internal:my-action/stream",
    ThreadPool.Names.SEARCH,
    MyRequest::new,
    this::handleStreamRequest
);

Basic Implementation Copy Chapter

private void handleStreamRequest(MyRequest request, TransportChannel channel, Task task) {
    try {
        // Process data incrementally
        DataIterator iterator = createDataIterator(request);
        
        while (iterator.hasNext()) {
            MyData data = iterator.next();
            MyResponse response = processData(data);
            
            // Send batch - may block or throw StreamException with CANCELLED code
            channel.sendResponseBatch(response);
        }
        
        // Signal successful completion
        channel.completeStream();
        
    } catch (StreamException e) {
        if (e.getErrorCode() == StreamErrorCode.CANCELLED) {
            // Client cancelled - exit gracefully
            logger.info("Stream cancelled by client: {}", e.getMessage());
            // Do NOT call completeStream() or sendResponse()
        } else {
            // Other stream error - send to client
            channel.sendResponse(e);
        }
        
    } catch (Exception e) {
        // Send error to client
        channel.sendResponse(e);
    }
}

Processing Flow Copy Chapter

flowchart TD
    A[Request Received] --> B[Process Data Loop]
    B --> C[Send Response Batch]
    C --> D{Client Cancelled?}
    D -->|Yes| E[Exit Gracefully]
    D -->|No| F{More Data?}
    F -->|Yes| B
    F -->|No| G[Complete Stream]
    G --> H[Success]
    
    B --> I{Error?}
    I -->|Yes| J[Send Error]
    J --> K[Terminated]
    
    classDef success fill:#e8f5e8
    classDef error fill:#ffebee
    classDef cancel fill:#fce4ec
    
    class G,H success
    class J,K error
    class E cancel

Key Behaviors Copy Chapter

Blocking Copy Section

• sendResponseBatch() may block if transport buffers are full
• Server will pause until client consumes data and frees buffer space
• See backpressure.md for behaviour, settings, and sizing.

Cancellation Copy Section

• sendResponseBatch() throws StreamException with StreamErrorCode.CANCELLED when client cancels
• Exit handler immediately - framework handles cleanup
• Do NOT call completeStream() or sendResponse() after cancellation

Completion Copy Section

• Always call either completeStream() (success) OR sendResponse(exception) (error)
• Never call both methods
• Stream must be explicitly completed or terminated

Trailing / per-batch metadata (native Arrow path) Copy Section

For actions whose response extends ArrowBatchResponse, opaque application metadata can be
attached to any batch via new MyResponse(root, byte[]) and read on the receiver via
response.getMetadata(). Attach to the last batch for stream-terminal payloads
(profiling counters, summary stats). See
native-arrow-transport-design.md.

Plugins Arrow Flight Rpc Docs Transport Client Streaming Flow (`plugins/arrow-flight-rpc/docs/transport-client-streaming-flow.md`) Copy Section

Client-Side Streaming API Flow

flowchart TD
    %% Simple Client Flow
    START[Client sends streaming request] --> WAIT[Wait for response]
    
    WAIT --> RESPONSE{Response Type?}
    RESPONSE -->|Success| STREAM[handleStreamResponse called]
    RESPONSE -->|Error| ERROR[handleException called]
    RESPONSE -->|Timeout| TIMEOUT[Timeout exception]
    
    %% Stream Processing
    STREAM --> NEXT[Get next response]
    NEXT --> PROCESS[Process response]
    PROCESS --> CONTINUE{Continue?}
    CONTINUE -->|Yes| NEXT
    CONTINUE -->|No - Complete| CLOSE[streamResponse.close]
    CONTINUE -->|No - Cancel| CANCEL[streamResponse.cancel]
    
    %% Error & Completion
    ERROR --> HANDLE_ERROR[Handle error]
    TIMEOUT --> HANDLE_ERROR
    CLOSE --> SUCCESS[Complete]
    CANCEL --> SUCCESS
    HANDLE_ERROR --> SUCCESS
    
    %% Simple styling
    classDef client fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px
    classDef framework fill:#e3f2fd,stroke:#1976d2,stroke-width:2px
    classDef error fill:#ffebee,stroke:#c62828,stroke-width:2px
    
    class START,NEXT,PROCESS,CLOSE,CANCEL client
    class WAIT,STREAM,ERROR,TIMEOUT framework
    class HANDLE_ERROR error
    class RESPONSE,CONTINUE decision

Simple Client Usage Copy Chapter

**Thread-Safe Implementation**: Copy Section

/* Detailed source-code truncated for AI context efficiency. */

**Key Points**: Copy Section

• Blocking: nextResponse() blocks waiting for server data - use background threads
• Timeout Handling: handleException can cancel active streams for timeout scenarios
• Always Close/Cancel: Stream must be closed or cancelled to prevent resource leaks

--- METRICS ---

• Files Extracted: 13
• Estimated Token Budget: ~18232 tokens
• Canonical Reference: https://codewiki.google/github.com/opensearch-project/OpenSearch