# Technical Documentation: opensearch-project/OpenSearch > ℹ️ **Provenance:** Multi-Tier Ingestion · [CodeWiki Reference](https://codewiki.google/github.com/opensearch-project/OpenSearch) ### Overview (README.md) [](https://github.com/opensearch-project/OpenSearch/blob/main/LICENSE.txt) [](https://insights.linuxfoundation.org/project/opensearch-foundation) [](https://insights.linuxfoundation.org/project/opensearch-foundation) [](https://www.bestpractices.dev/projects/6556) [](https://codecov.io/gh/opensearch-project/OpenSearch) [](https://www.linkedin.com/company/opensearch-project) - [Welcome!](#welcome) - [Project Resources](#project-resources) - [Code of Conduct](#code-of-conduct) - [Security](#security) - [License](#license) - [Copyright](#copyright) - [Trademark](#trademark) ## Welcome! OpenSearch is an open-source, enterprise-grade search and observability suite that brings order to unstructured data at scale. ## Project Resources * [Project Website](https://opensearch.org/) * [Downloads](https://opensearch.org/downloads/) * [Documentation](https://docs.opensearch.org/) * Need help? Try [Forums](https://discuss.opendistrocommunity.dev/) or [Slack](https://opensearch.org/slack/) * [Contributing to OpenSearch](CONTRIBUTING.md) * [Maintainer Responsibilities](MAINTAINERS.md) * [Release Management](RELEASING.md) * [Admin Responsibilities](ADMINS.md) * [Testing](TESTING.md) * [Security](SECURITY.md) ## Code of Conduct The project's [Code of Conduct](CODE_OF_CONDUCT.md) outlines our expectations for all participants in our community, based on the [OpenSearch Code of Conduct](https://opensearch.org/code-of-conduct/). Please contact [conduct@opensearch.foundation](mailto:conduct@opensearch.foundation) with any additional questions or comments. ## Security If you discover a potential security issue in this project we ask that you notify OpenSearch Security directly via email to security@opensearch.org. Please do **not** create a public GitHub issue. ## License This project is licensed under the [Apache v2.0 License](LICENSE.txt). ## Copyright Copyright OpenSearch Contributors. See [NOTICE](NOTICE.txt) for details. ## Trademark 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`) # 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`) # 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 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 `BatchTask`s; 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 | Setting | Default | Property | |---|---|---| | `arrow.flight.channel.ready_timeout` | `60s` | node-scope | | `arrow.flight.channel.outbound_buffer_threshold` | `64mb` | node-scope | - `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 ### Fast consumer (steady state) The readiness check is essentially free: the producer's call to `awaitReadyOrThrow` returns immediately because gRPC's outbound buffer stays below threshold. ### Slow consumer - `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 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 ### Sizing the flight pool `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 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 `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) 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 > **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`) # FlightServerChannel Lifecycle & Buffer-Ownership Contract ## Purpose 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 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). | Name | What it is | What it runs | |------|------------|--------------| | **`T_caller`** | The producer's calling thread (e.g. a `SEARCH` worker) inside `FlightOutboundHandler.sendResponseBatch`. | `awaitReadyOrThrow()` (the back-pressure gate) and `executor.execute(task)`. **This is where a batch is built but not yet handed to the channel.** | | **`T_exec`** | The channel's **single-thread** flight executor (`FlightTransport.getNextFlightExecutor()`). | Everything the handler submits: `processBatchTask`→`sendBatch`, `processCompleteTask`→`completeStream`, `processErrorTask`→`sendError`, `failStream`. Also `ArrowFlightProducer.getStream`'s bootstrap. | | **`T_grpc`** | gRPC's cancel-delivery thread (`cancelExecutor` in `JumpToApplicationThreadServerStreamListener`). | `onChannelCancelled()` → `close()`. | **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 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 | Component | Role | Owns | MUST NOT | |-----------|------|------|----------| | **`FlightServerChannel`** | **Sole owner** of all Arrow buffers and the gRPC stream lifecycle. The only place a root is freed and the only place a terminal gRPC op is issued. | stream `root`, gRPC listener, `terminalSent`, close listeners | leak/double-free a root; issue two terminal ops; act on the stream after terminal or after `cancelled` | | **`FlightOutboundHandler`** | **Stateless translator/dispatcher.** Thread-switch onto `T_exec`, back-pressure gate, translate response→channel call, relay outcome. Owns the **hand-off discipline** (§3) for the source root until the channel takes it. | nothing persistent | touch/free the *stream* root; call `close()` directly; serialize a response itself | | **`FlightTransportChannel`** | Thin framework adapter. `release()` → `channel.close()`. | nothing | contain its own close/idempotency logic | | **`ArrowFlightProducer`** | Per-call bootstrap. May `close()` on early/inbound errors. | nothing | assume "double close is harmless" without the channel guaranteeing it | | **`T_grpc` (cancel)** | Signals cancellation → `close()`. | — | — | **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 | Buffer | Created by | Freed by | When | |--------|-----------|----------|------| | **Source root** — `ArrowBatchResponse.getRoot()`, the producer's per-batch native batch | the caller/producer | **the channel** — via `sendBatch`'s `finally` if handed off, else via `channel.releaseUnsent(response)` on the failed-handoff path (§3.1) | exactly once | | **Stream root** — the single reused VSR (native *or* byte) sent on the wire | the channel, lazily on the first batch | **the channel**, in `close()` only | terminal, exactly once | | **Metadata buf** — per-batch app metadata on the Flight frame | the channel, per batch | gRPC via `putNext(ArrowBuf)`, or the channel if `putNext` throws before hand-off (§3.3) | handoff | | **Byte-serialization vector** — the `VarBinary` vector of the byte stream root | the channel (it *is* the stream root's vector) | **the channel**, in `close()` | terminal, exactly once — **never per-batch** (§3.2) | **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) 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 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 ``` 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) ``` 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` `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. ```java 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> 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) 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 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 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 ``` 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 ``` 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) ``` if (response instanceof ArrowBatchResponse r && r.getRoot() != null) r.getRoot().close(); // never touches stream root, terminalSent, or open ``` ### `close()` — fire-once, idempotent, ANY thread ``` 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` ``` 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 ``` 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` ``` if (cancelled) return; cancelled = true; recordCallEnd(CANCELLED); close(); ``` --- ## 6. Scenario walk-through `✓` = 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) 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) 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 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.) | # | Root-cause class | Sev | Bug type | Closed by | |---|------------------|-----|----------|-----------| | 1 | Non-atomic `close()` double-entry → double-free root + double `notifyCloseListeners` (crash) | blocker | crash/double-free | `open.compareAndSet` (inv 1,2); S5 | | 2 | Cancel frees stream root mid-`transferRoot`/`putNext` (UAF) | blocker | UAF | single-writer: root free posted to `T_exec`, behind any send (inv 4); S4 | | 3 | Cancel after transfer-into-reused-root, send rejects → transferred buffers leak; `root` never nulled | blocker | leak/UAF | gate **before** transfer + null-after-free; free serialized on `T_exec` (inv 2,4); S4 | | 4 | Source root orphaned when send never runs: gate throw / `execute` reject / wrong-channel return | blocker | leak | `handedOff` + `releaseUnsent` (§3.1, inv 5); S7 | | 5 | Source root orphaned inside `sendBatch`: `getHeaderBuffer`/`create`/`transferRoot` throw | major | leak | header built before transfer + `finally` frees source (§3.4, §5); S8 | | 6 | Byte path: per-batch `out.close()` frees reused vector (UAF vs zero-copy & vs cancel) | blocker | UAF | channel owns vector, freed once in `close()` (§3.2); S9 | | 7 | Terminal-after-cancel: `completeStream`/`sendError` issue terminal op after cancel | major | illegal transition | `cancelled` guard (inv 3) | | 8 | Metadata buf leaked when `putNext` throws before hand-off | minor | leak | try/catch + `refCnt` close (§3.3) | | 9 | `addCloseListener` races `close()` flip+clear → lost fire / CME / tracker-map leak | major | leak/visibility | atomic registration (inv 7, §5) | | 10 | Throwing close listener strands remaining listeners (TaskManager untrack skipped) | minor | leak | per-listener try/catch (inv 8, §5) | | 11 | `shutdownNow` drops queued tasks undrained → source roots leak | minor | leak (shutdown-only) | **documented limitation** — node teardown; OS reclaims on exit (§10) | | 12 | Wrong-channel-type / NOOP-listener early returns (defensive `instanceof`) | info | n/a | covered by §3.1 handoff (handedOff stays false → `releaseUnsent`) | | 13 | Non-`Exception` `Throwable` (OOM / `AssertionError`) from `start`/`putNext`/transfer **after** stream-root adoption → `processBatchTask` only caught `Exception`, so the channel was never released → adopted stream root leaks | major | leak | `catch (Throwable)` in `processBatchTask` routes through `failStream` (releases channel → `close()` frees the root) + notifies the listener; `failStream` closes the channel directly when there is no transport channel (§3.5); S10 | --- ## 10. Known limitations (not closed by this model) - **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 `BatchTask`s — 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 Add this task to create the chaos agent JAR: ```gradle // 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 Once enabled, run the chaos tests with: ```bash ./gradlew :plugins:arrow-flight-rpc:internalClusterTest --tests="*Chaos*" ``` ## What Chaos Testing Does 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`) # Arrow Flight RPC Error Handling Guidelines ## Overview 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 The following error codes are available in `StreamErrorCode`: | StreamErrorCode | Description | |--------------------|------------------------------------------------------------| | OK | Operation completed successfully | | CANCELLED | Operation was cancelled by the client | | UNKNOWN | Unknown error or unhandled server exception | | INVALID_ARGUMENT | Invalid arguments provided | | TIMED_OUT | Operation timed out | | NOT_FOUND | Requested resource not found | | ALREADY_EXISTS | Resource already exists | | UNAUTHENTICATED | Client not authenticated | | UNAUTHORIZED | Client lacks permission for the operation | | RESOURCE_EXHAUSTED | Resource limits exceeded | | UNIMPLEMENTED | Operation not implemented | | INTERNAL | Internal server error | | UNAVAILABLE | Service unavailable or resource temporarily inaccessible | ## Best Practices ### Throwing Errors When throwing errors in server-side code: ```java // 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 When handling errors in client-side code: ```java 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 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()` ```java 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 `StreamException` supports adding metadata for additional error context: ```java 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: ```java Map 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`) # 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`) # 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 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 Streaming transport tasks can be monitored using the existing Tasks API: ```bash 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 Metrics are organized into the following categories: ### Client Call Metrics Metrics related to client-side calls: | Metric | Description | |--------|-------------| | `started` | Number of client calls started | | `completed` | Number of client calls completed | | `duration` | Duration statistics for client calls (min, max, avg, sum) | | `request_bytes` | Size statistics for requests sent by clients (min, max, avg, sum) | | `response` | Total size of responses received by clients (with human-readable format) | | `response_bytes` | Total size of responses received by clients (in bytes) | ### Client Batch Metrics Metrics related to client-side batch operations: | Metric | Description | |--------|-------------| | `requested` | Number of batches requested by clients | | `received` | Number of batches received by clients | | `received_bytes` | Size statistics for batches received by clients (min, max, avg, sum) | | `processing_time` | Time statistics for processing received batches (min, max, avg, sum) | ### Server Call Metrics Metrics related to server-side calls: | Metric | Description | |--------|-------------| | `started` | Number of server calls started | | `completed` | Number of server calls completed | | `duration` | Duration statistics for server calls (min, max, avg, sum) | | `request_bytes` | Size statistics for requests received by servers (min, max, avg, sum) | | `response` | Total size of responses sent by servers (with human-readable format) | | `response_bytes` | Total size of responses sent by servers (in bytes) | ### Server Batch Metrics Metrics related to server-side batch operations: | Metric | Description | |--------|-------------| | `sent` | Number of batches sent by servers | | `sent_bytes` | Size statistics for batches sent by servers (min, max, avg, sum) | | `processing_time` | Time statistics for processing and sending batches (min, max, avg, sum) | ### Status Metrics Metrics related to call status codes: | Metric | Description | |--------|-------------| | `client.{status}` | Count of client calls completed with each status code (OK, CANCELLED, UNAVAILABLE, etc.) | | `server.{status}` | Count of server calls completed with each status code (OK, CANCELLED, UNAVAILABLE, etc.) | ### Resource Metrics Metrics related to resource usage: | Metric | Description | |--------|-------------| | `arrow_allocated` | Current Arrow memory allocation (human-readable format) | | `arrow_allocated_bytes` | Current Arrow memory allocation in bytes | | `arrow_peak` | Peak Arrow memory allocation (human-readable format) | | `arrow_peak_bytes` | Peak Arrow memory allocation in bytes | | `direct_memory` | Current direct memory usage (human-readable format) | | `direct_memory_bytes` | Current direct memory usage in bytes | | `client_threads_active` | Number of active client threads | | `client_threads_total` | Total number of client threads | | `server_threads_active` | Number of active server threads | | `server_threads_total` | Total number of server threads | | `client_channels_active` | Number of active client channels | | `server_channels_active` | Number of active server channels | ## Cluster-Level Metrics 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 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## Interpreting Metrics ### Performance Monitoring - **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 - **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 - **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`) # Native Arrow Transport Path ## Overview 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 ### 1. Define your response Extend `ArrowBatchResponse`. No `writeTo`/`read` override needed — the framework handles it. ```java 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 ```java 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 ```java class MyQueryHandler implements StreamTransportResponseHandler { public MyQueryResponse read(StreamInput in) throws IOException { return new MyQueryResponse(in); } public void handleStreamResponse(StreamTransportResponse 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 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 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 | Resource | Created by | Closed by | |----------|-----------|-----------| | Channel allocator | Framework | Framework (on channel close) | | Producer root (per batch) | Producer | Framework (after zero-copy transfer on executor) | | Shared Flight root | Framework | Framework (on channel close) | 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 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 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. ```java // 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`) # 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 ```mermaid 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 ```mermaid 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 ```mermaid 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 ```mermaid 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 TransportResponseHandler
using reqID FCC->>RH2: handler.handleStreamResponse(streamResponse)
(Async Processing) ``` ## Key Differences Summary ### **Netty4 Transport (Traditional)**: - **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)**: - **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`) # 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](#transport-modes-flight-vs-flight-secure) for details. ## Configuration Add these settings to `opensearch.yml`: ```yaml # 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 The plugin provides two transport modes. Choosing the correct mode is critical for security: | Mode | TLS | Use Case | |------|-----|----------| | `FLIGHT-SECURE` | Encrypted (mTLS) | **Production — always use this when security is enabled** | | `FLIGHT` | Plaintext | Development/testing only, with no security plugin | ### FLIGHT-SECURE (Recommended) `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) > **⚠️ 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 Install and configure the security plugin: ```bash # Install security plugin bin/opensearch-plugin install opensearch-security # Setup demo configuration plugins/opensearch-security/tools/install_demo_configuration.sh ``` ## Secure Configuration Checklist When deploying with the security plugin, verify all three settings are present: ```yaml # 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 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 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`) # Server-Side Streaming API Guide ## Overview 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 ```java streamTransportService.registerRequestHandler( "internal:my-action/stream", ThreadPool.Names.SEARCH, MyRequest::new, this::handleStreamRequest ); ``` ## Basic Implementation ```java 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 ```mermaid 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 ### Blocking - `sendResponseBatch()` may block if transport buffers are full - Server will pause until client consumes data and frees buffer space - See [backpressure.md](backpressure.md) for behaviour, settings, and sizing. ### Cancellation - `sendResponseBatch()` throws `StreamException` with `StreamErrorCode.CANCELLED` when client cancels - Exit handler immediately - framework handles cleanup - Do NOT call `completeStream()` or `sendResponse()` after cancellation ### Completion - 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) 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](native-arrow-transport-design.md#application-metadata). --- ### Plugins Arrow Flight Rpc Docs Transport Client Streaming Flow (`plugins/arrow-flight-rpc/docs/transport-client-streaming-flow.md`) # Client-Side Streaming API Flow ```mermaid 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 ### **Thread-Safe Implementation**: ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### **Key Points**: - **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