Skip root transfer for byte-serialized via handler marker

[bowenlan-amzn]

Description

On the receiving side of stream transport, FlightTransportResponse.nextResponse() transfers the Flight stream root into a response-owned root on every batch — but byte-serialized responses copy bytes into Java fields via handler.read() and never touch the vectors. The transfer is wasted work on that path.

This PR splits VectorStreamInput into two subclasses and let the handler decide which path to take.

flowchart LR
    stream["FlightStream<br/>(stream root)"]
    dispatch{"handler.skips<br/>Deserialization()?"}
    byte["ByteSerialized<br/>reads bytes from stream root"]
    native["NativeArrow<br/>transfers into consumer root"]

    stream --> dispatch
    dispatch -- false --> byte
    dispatch -- true --> native

Loading

Handler marker

TransportResponseHandler gains a skipsDeserialization() bit (default false). ArrowBatchResponseHandler pins it to true. Wrappers forward their delegate's value:

flowchart LR
    M["MetricsTracking"] --> C["ContextRestore"] --> T["Traceable<br/>(optional)"] --> U["User handler"]

    M -. "skipsDeserialization()" .-> U

Loading

Ownership lifecycle of Arrow on the Receive End

sequenceDiagram
    participant FS as FlightStream
    participant FTR as FlightTransportResponse
    participant VSI as VectorStreamInput.NativeArrow
    participant ABR as ArrowBatchResponse

    FTR->>FS: getRoot()
    FS-->>FTR: stream root
    FTR->>VSI: newStreamInput(streamRoot)
    Note over VSI: transfer to consumer
    FTR->>ABR: handler.read(input)
    ABR->>VSI: getRoot() + claimOwnership()
    Note over ABR: response now owns the root
    FTR->>VSI: close()
    Note over VSI: no-op (ownership claimed)

Loading

Shared allocator

ArrowAllocatorProvider provides a single node-level RootAllocator so all Arrow plugins share the same root — required for zero-copy transfers to pass Arrow's AllocationManager associate check.

flowchart TD
    root["ArrowAllocatorProvider.ROOT"]
    flight["flight"]
    server["server"]
    client["client"]
    plugin["plugin allocators<br/>(e.g. analytics-search)"]

    root --> flight
    root --> plugin
    flight --> server
    flight --> client

Loading

Who should close the Arrow resource

Send: arrow batch is transfered from producer to flight, so flight should close
Receive: arrow batch is transfered from flight to consumer, so consumer should close

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 6a09343)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Add skipsDeserialization() marker to TransportResponseHandler and forwarding wrappers Relevant files: server/src/main/java/org/opensearch/transport/TransportResponseHandler.java server/src/main/java/org/opensearch/transport/TransportService.java server/src/main/java/org/opensearch/telemetry/tracing/handler/TraceableTransportResponseHandler.java plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/MetricsTrackingResponseHandler.java plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowBatchResponseHandler.java Sub-PR theme: Introduce shared ArrowAllocatorProvider and migrate allocator injection Relevant files: plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java plugins/examples/stream-transport-example/src/main/java/org/opensearch/example/stream/ExampleAllocator.java plugins/examples/stream-transport-example/src/main/java/org/opensearch/example/stream/StreamTransportExamplePlugin.java plugins/examples/stream-transport-example/src/main/java/org/opensearch/example/stream/TransportNativeArrowStreamDataAction.java Sub-PR theme: Split VectorStreamInput into ByteSerialized/NativeArrow and skip root transfer for byte path Relevant files: plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowBatchResponse.java plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransportResponse.java plugins/arrow-flight-rpc/src/test/java/org/opensearch/arrow/flight/transport/VectorStreamInputTests.java plugins/arrow-flight-rpc/src/test/java/org/opensearch/arrow/flight/transport/ArrowBatchResponseTests.java

⚡ Recommended focus areas for review Singleton Lifecycle The static ROOT RootAllocator is initialized at class-load time and never closed. If multiple test JVMs or plugin reloads occur, the allocator leaks. There is no shutdown hook or close mechanism for the root allocator, which could cause memory accounting errors in Arrow. private static final RootAllocator ROOT = AccessController.doPrivileged( (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) ); Null Root Risk The send-side constructor accepts a null VectorSchemaRoot (as seen in the test TestArrowResponse() which passes null). If getRoot() is called on a send-side instance that was constructed with null, downstream code (e.g., FlightOutboundHandler accessing fieldVectors) will throw a NullPointerException. Consider adding a null check or documenting that null is only valid in specific test scenarios. protected ArrowBatchResponse(VectorSchemaRoot batchRoot) { this.batchRoot = batchRoot; } Missing Transfer on Native Path On the native Arrow path, forNativeArrow() transfers vectors from the stream root into a consumer root. However, after handler.read() returns (which calls claimOwnership), the stream root is still held by FlightStream. On the next flightStream.next() call, FlightStream will clear/overwrite the stream root. Since ownership was transferred, this should be safe — but it should be validated that the consumer root truly holds independent memory after the transfer and that no reference to the stream root escapes into the response. VectorSchemaRoot streamRoot = flightStream.getRoot(); currentBatchSize = FlightUtils.calculateVectorSchemaRootSize(streamRoot); try (VectorStreamInput input = newStreamInput(streamRoot)) { input.setVersion(initialHeader.getVersion()); return handler.read(input); } Resource Leak on Exception In processBatchTask, when streamRoot is null and a new VectorSchemaRoot is created, if FlightUtils.transferRoot() or subsequent operations throw, the newly created streamRoot is not closed. The old code had the same issue but the new code should add a try-finally or try-with-resources to ensure the new streamRoot is closed on error. VectorSchemaRoot streamRoot = flightChannel.getRoot(); if (streamRoot == null) { // Create stream root using the producer's allocator for same-allocator transfer. // This avoids an Arrow bug where cross-allocator transferOwnership of foreign-backed // buffers (from C data import) doesn't properly free the ArrowArray C struct. // The producer's allocator must be long-lived (not closed per-request). List<FieldVector> fieldVectors = arrowResponse.getRoot().getFieldVectors(); if (fieldVectors.isEmpty()) { throw new IllegalStateException("Native Arrow batch has no field vectors"); } streamRoot = VectorSchemaRoot.create(arrowResponse.getRoot().getSchema(), fieldVectors.getFirst().getAllocator()); } FlightUtils.transferRoot(arrowResponse.getRoot(), streamRoot); arrowResponse.getRoot().close(); out = VectorStreamOutput.forNativeArrow(streamRoot); Thread Safety The NativeArrow.transferred flag is not volatile or synchronized. If claimOwnership() and close() are called from different threads (e.g., handler.read() on one thread and try-with-resources cleanup on another), there is a potential race condition where close() may still release the root after claimOwnership() was called. private boolean transferred = false; NativeArrow(VectorSchemaRoot root, NamedWriteableRegistry registry) { super(root, registry); } @Override public byte readByte() { throw new UnsupportedOperationException("Native Arrow responses read vectors directly from getRoot()"); } @Override public void readBytes(byte[] b, int offset, int len) { throw new UnsupportedOperationException("Native Arrow responses read vectors directly from getRoot()"); } /** Response claims the consumer root; {@link #close()} becomes a no-op. */ void claimOwnership() { transferred = true; } /** Releases the consumer root unless {@link #claimOwnership()} was called. */ @Override public void close() { if (!transferred && root != null) { root.close(); } }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 6a09343

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix inconsistent allocator limit causing silent memory cap ArrowAllocatorProvider.newChildAllocator accepts a long limit, but Integer.MAX_VALUE is passed here while the root allocator uses Long.MAX_VALUE. This inconsistency silently caps the flight allocator at ~2 GB. Use Long.MAX_VALUE to match the root allocator's limit. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java [146-148] -flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Integer.MAX_VALUE); +flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Long.MAX_VALUE); serverAllocator = flightAllocator.newChildAllocator("server", 0, flightAllocator.getLimit()); clientAllocator = flightAllocator.newChildAllocator("client", 0, flightAllocator.getLimit()); Suggestion importance[1-10]: 7 __ Why: The flightAllocator is created with Integer.MAX_VALUE (~2 GB) while the root allocator uses Long.MAX_VALUE. This silently caps the flight transport memory at ~2 GB, which is a real functional inconsistency that could cause unexpected allocation failures in production.	Medium
General	Add shutdown hook to release static root allocator The ROOT allocator is a static singleton that is never closed, which will cause resource leaks in tests and environments where the class is loaded multiple times. Consider adding a JVM shutdown hook or providing an explicit close() method to release the root allocator when the application shuts down. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java [32-34] private static final RootAllocator ROOT = AccessController.doPrivileged( (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) ); +static { + Runtime.getRuntime().addShutdownHook(new Thread(ROOT::close, "arrow-root-allocator-shutdown")); +} + Suggestion importance[1-10]: 4 __ Why: The static ROOT allocator is never closed, which could cause resource leaks. However, adding a JVM shutdown hook for this is a debatable pattern — the OS reclaims memory on JVM exit anyway, and shutdown hooks can cause issues in test environments. The suggestion is valid but has limited practical impact.	Low
	Add null guard for send-side constructor argument The null check in == null ? "null" : in.getClass().getName() in the error message is unreachable because in instanceof VectorStreamInput.NativeArrow already handles the null case (instanceof returns false for null). More importantly, if batchRoot is null (e.g., when the send-side constructor is called with null), getRoot() will return null and callers will get a NullPointerException without a clear message. Consider adding a null guard in the send-side constructor. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowBatchResponse.java [82-95] +protected ArrowBatchResponse(VectorSchemaRoot batchRoot) { + if (batchRoot == null) { + throw new IllegalArgumentException("batchRoot must not be null on the send side"); + } + this.batchRoot = batchRoot; +} + protected ArrowBatchResponse(StreamInput in) throws IOException { super(in); if (in instanceof VectorStreamInput.NativeArrow nativeIn) { this.batchRoot = nativeIn.getRoot(); nativeIn.claimOwnership(); } else { throw new IllegalStateException( "ArrowBatchResponse decoded from a non-native-Arrow StreamInput (" - + (in == null ? "null" : in.getClass().getName()) + + in.getClass().getName() + "). Wrapping handlers around ArrowBatchResponseHandler must forward " + "TransportResponseHandler#skipsDeserialization()." ); } } Suggestion importance[1-10]: 3 __ Why: The null check in the error message is indeed unreachable (instanceof handles null), and the suggestion to clean it up is valid. However, adding a null guard to the send-side constructor is a minor defensive improvement, and the test code in FlightTransportResponseTests explicitly passes null to the send-side constructor, suggesting null may be intentionally allowed.	Low
	Verify transfer safety with FlightStream lifecycle After FlightUtils.transferRoot drains streamRoot, the streamRoot's row count is set to 0 but the vectors are still allocated. If the caller (FlightStream) later calls close() on streamRoot, it will try to release already-transferred buffers, potentially causing double-free errors. The transfer should be verified to be safe with the FlightStream lifecycle, or the stream root should be explicitly cleared after transfer. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java [64-79] +static VectorStreamInput forNativeArrow(VectorSchemaRoot streamRoot, NamedWriteableRegistry registry) { + if (streamRoot.getFieldVectors().isEmpty()) { + throw new IllegalStateException("Native Arrow batch has no field vectors"); + } + VectorSchemaRoot consumerRoot = VectorSchemaRoot.create( + streamRoot.getSchema(), + streamRoot.getFieldVectors().getFirst().getAllocator() + ); + try { + FlightUtils.transferRoot(streamRoot, consumerRoot); + } catch (Throwable t) { + consumerRoot.close(); + throw t; + } + return new NativeArrow(consumerRoot, registry); +} - Suggestion importance[1-10]: 2 __ Why: The suggestion asks to verify behavior rather than proposing a concrete fix, and the existing_code equals the improved_code. Arrow's transferOwnership moves buffer ownership so double-free is not a concern — the transferred buffers are no longer owned by streamRoot.	Low

---

Previous suggestions

Suggestions up to commit 6a09343

Category	Suggestion	Impact
Possible issue	Fix inconsistent allocator limit from int to long ArrowAllocatorProvider.newChildAllocator accepts a long limit, but Integer.MAX_VALUE is passed here while the root allocator uses Long.MAX_VALUE. This inconsistency silently caps the flight allocator at ~2 GB. Pass Long.MAX_VALUE to be consistent with the root allocator's intent. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java [146-148] -flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Integer.MAX_VALUE); +flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Long.MAX_VALUE); serverAllocator = flightAllocator.newChildAllocator("server", 0, flightAllocator.getLimit()); clientAllocator = flightAllocator.newChildAllocator("client", 0, flightAllocator.getLimit()); Suggestion importance[1-10]: 6 __ Why: The flightAllocator is created with Integer.MAX_VALUE (~2 GB) while the root allocator uses Long.MAX_VALUE. This is a real inconsistency that silently caps memory usage, and the fix is straightforward — change to Long.MAX_VALUE.	Low
General	Add shutdown hook to close static root allocator The ROOT allocator is a static singleton that is never closed, which will cause resource leaks in tests or environments where the class is loaded multiple times. Consider adding a JVM shutdown hook or providing an explicit close() method to release the root allocator when the application shuts down. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java [32-34] private static final RootAllocator ROOT = AccessController.doPrivileged( (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) ); +static { + Runtime.getRuntime().addShutdownHook(new Thread(ROOT::close, "arrow-root-allocator-shutdown")); +} + Suggestion importance[1-10]: 4 __ Why: The static ROOT allocator is never closed, which could cause resource leaks. However, adding a JVM shutdown hook is a debatable approach for a library component, and the impact is limited since this is typically a long-lived singleton. The suggestion is valid but has moderate importance.	Low
	Handle suppressed exceptions during consumer root cleanup After FlightUtils.transferRoot drains streamRoot, the streamRoot's row count is set to 0 but its vectors are not closed — the caller (FlightStream) still owns them. However, if transferRoot partially transfers vectors before throwing, the already-transferred buffers in consumerRoot may be double-freed when consumerRoot.close() is called in the catch block. Verify that FlightUtils.transferRoot is atomic or add per-vector error handling to avoid double-free on partial transfer. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java [64-79] static VectorStreamInput forNativeArrow(VectorSchemaRoot streamRoot, NamedWriteableRegistry registry) { if (streamRoot.getFieldVectors().isEmpty()) { throw new IllegalStateException("Native Arrow batch has no field vectors"); } VectorSchemaRoot consumerRoot = VectorSchemaRoot.create( streamRoot.getSchema(), streamRoot.getFieldVectors().getFirst().getAllocator() ); try { FlightUtils.transferRoot(streamRoot, consumerRoot); + return new NativeArrow(consumerRoot, registry); } catch (Throwable t) { - consumerRoot.close(); + // Only close consumerRoot if transfer was not partial; otherwise log and rethrow. + try { + consumerRoot.close(); + } catch (Throwable suppressed) { + t.addSuppressed(suppressed); + } throw t; } - return new NativeArrow(consumerRoot, registry); } Suggestion importance[1-10]: 3 __ Why: The suggestion to add suppressed exception handling is a minor improvement for error reporting. The improved_code also moves return new NativeArrow(...) inside the try block, which is a minor structural improvement, but the core logic remains the same and the risk of double-free depends on FlightUtils.transferRoot internals not shown in the diff.	Low
	Lazily initialize allocator to prevent resource leaks The allocator field is initialized eagerly at field-declaration time, before close() is guaranteed to be called by the test framework. If plugin instantiation fails partway through, the allocator may be leaked. Consider initializing the allocator lazily in createComponents or in a try-finally block to ensure it is always closed. plugins/arrow-flight-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/NativeArrowTransportIT.java [504] public static class NativeArrowTestPlugin extends Plugin implements ActionPlugin { - private final BufferAllocator allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + private BufferAllocator allocator; + @Override + public Collection<Object> createComponents(...) { + allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + return List.of(new TestAllocatorHolder(allocator)); + } + + @Override + public void close() { + if (allocator != null) { + allocator.close(); + } + } + Suggestion importance[1-10]: 3 __ Why: The concern about eager initialization is valid but minor in a test context where plugin lifecycle is controlled by the test framework. The improved_code is incomplete (uses ... for parameters) and the risk of a leak from eager initialization in tests is low.	Low

Suggestions up to commit 6a09343

Category	Suggestion	Impact
General	Prevent static root allocator resource leak The ROOT allocator is a static singleton that is never closed, which will cause resource leaks in tests and environments where the class is loaded multiple times. Consider adding a JVM shutdown hook or providing an explicit close() method to release the root allocator when the application shuts down. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java [32-34] -private static final RootAllocator ROOT = AccessController.doPrivileged( - (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) -); +private static final RootAllocator ROOT; +static { + ROOT = AccessController.doPrivileged( + (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) + ); + Runtime.getRuntime().addShutdownHook(new Thread(ROOT::close, "arrow-root-allocator-shutdown")); +} Suggestion importance[1-10]: 5 __ Why: The static ROOT allocator is never closed, which could cause resource leaks. However, adding a JVM shutdown hook for a static singleton is a common pattern but may not be the best approach for a library/plugin context where lifecycle is managed externally. The suggestion is valid but has moderate impact.	Low
	Fix inconsistent allocator limit type The flightAllocator is created with Integer.MAX_VALUE as the limit, but ArrowAllocatorProvider.newChildAllocator accepts a long. This is inconsistent with the root allocator's Long.MAX_VALUE limit and may unnecessarily cap memory. Additionally, flightAllocator.getLimit() returns the child's limit (Integer.MAX_VALUE), so serverAllocator and clientAllocator are each capped at Integer.MAX_VALUE independently, which may not be the intended behavior. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java [146-148] -flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Integer.MAX_VALUE); +flightAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Long.MAX_VALUE); serverAllocator = flightAllocator.newChildAllocator("server", 0, flightAllocator.getLimit()); clientAllocator = flightAllocator.newChildAllocator("client", 0, flightAllocator.getLimit()); Suggestion importance[1-10]: 5 __ Why: Using Integer.MAX_VALUE instead of Long.MAX_VALUE for the flightAllocator limit is inconsistent with the root allocator's Long.MAX_VALUE and unnecessarily caps memory at ~2GB. This is a valid correctness concern worth addressing.	Low
	Remove unreachable null check in error message The null check (in == null ? "null" : in.getClass().getName()) in the error message is unreachable: if in were null, the instanceof pattern match would have already evaluated to false without a NullPointerException, but super(in) would have thrown before reaching this code. The null guard is misleading and can be simplified. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowBatchResponse.java [82-95] protected ArrowBatchResponse(StreamInput in) throws IOException { super(in); if (in instanceof VectorStreamInput.NativeArrow nativeIn) { this.batchRoot = nativeIn.getRoot(); nativeIn.claimOwnership(); } else { throw new IllegalStateException( "ArrowBatchResponse decoded from a non-native-Arrow StreamInput (" - + (in == null ? "null" : in.getClass().getName()) + + in.getClass().getName() + "). Wrapping handlers around ArrowBatchResponseHandler must forward " + "TransportResponseHandler#skipsDeserialization()." ); } } Suggestion importance[1-10]: 3 __ Why: The null check (in == null ? "null" : in.getClass().getName()) is indeed unreachable since super(in) would throw before reaching the else branch if in were null. Removing it improves code clarity, though the impact is minor.	Low
	Document partial transfer failure behavior After FlightUtils.transferRoot succeeds, the streamRoot is drained (its vectors are moved to consumerRoot). However, the caller (FlightTransportResponse.nextResponse) still holds a reference to streamRoot (the FlightStream's root), and the next call to flightStream.next() will reuse it. This is correct, but if transferRoot only partially transfers vectors before throwing, the streamRoot may be left in a partially drained state. Ensure FlightUtils.transferRoot is atomic or document that partial failure leaves streamRoot in an undefined state. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java [64-79] static VectorStreamInput forNativeArrow(VectorSchemaRoot streamRoot, NamedWriteableRegistry registry) { if (streamRoot.getFieldVectors().isEmpty()) { throw new IllegalStateException("Native Arrow batch has no field vectors"); } VectorSchemaRoot consumerRoot = VectorSchemaRoot.create( streamRoot.getSchema(), streamRoot.getFieldVectors().getFirst().getAllocator() ); try { FlightUtils.transferRoot(streamRoot, consumerRoot); } catch (Throwable t) { consumerRoot.close(); + // streamRoot may be partially drained; caller should treat it as invalid throw t; } return new NativeArrow(consumerRoot, registry); } Suggestion importance[1-10]: 2 __ Why: The suggestion only adds a comment to document behavior without changing any logic. The improved_code is functionally identical to the existing_code, making this a documentation-only change with minimal impact.	Low

Suggestions up to commit d933160

Category	Suggestion	Impact
Possible issue	Guard against empty field vectors before index access If arrowResponse.getRoot().getFieldVectors() is empty (e.g., a schema with no columns), get(0) will throw an IndexOutOfBoundsException. This is the same guard that was added in VectorStreamInput.forNativeArrow. A null/empty check should be added here as well to prevent an unhandled exception on the executor thread. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightOutboundHandler.java [162-165] +List<org.apache.arrow.vector.FieldVector> fieldVectors = arrowResponse.getRoot().getFieldVectors(); +if (fieldVectors.isEmpty()) { + throw new IllegalStateException("ArrowBatchResponse has no field vectors"); +} streamRoot = VectorSchemaRoot.create( arrowResponse.getRoot().getSchema(), - arrowResponse.getRoot().getFieldVectors().get(0).getAllocator() + fieldVectors.get(0).getAllocator() ); Suggestion importance[1-10]: 6 __ Why: The get(0) call on getFieldVectors() can throw IndexOutOfBoundsException if the schema has no columns, and VectorStreamInput.forNativeArrow already has this guard. Adding the same protection in FlightOutboundHandler is a valid consistency and safety improvement.	Low
	Avoid closing shared allocator in transport teardown rootAllocator is now a child allocator from ArrowAllocatorProvider, not a RootAllocator. Calling rootAllocator.getLimit() on a child allocator returns the child's own limit, which is fine, but rootAllocator.newChildAllocator(...) creates grandchildren of the shared root. More critically, when FlightTransport is closed and calls rootAllocator.close(), it closes a shared child allocator that other plugins may still be using, potentially causing memory accounting errors. The FlightTransport should not close the shared root's child if it doesn't own the root lifecycle. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java [146-148] rootAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Long.MAX_VALUE); -serverAllocator = rootAllocator.newChildAllocator("server", 0, rootAllocator.getLimit()); -clientAllocator = rootAllocator.newChildAllocator("client", 0, rootAllocator.getLimit()); +serverAllocator = ArrowAllocatorProvider.newChildAllocator("flight-server", Long.MAX_VALUE); +clientAllocator = ArrowAllocatorProvider.newChildAllocator("flight-client", Long.MAX_VALUE); Suggestion importance[1-10]: 5 __ Why: The concern about closing a shared child allocator is valid in principle, but the improved_code changes the hierarchy by creating flat siblings instead of nested children, which changes the memory accounting structure. The suggestion raises a real lifecycle concern but the fix may not be the right approach without understanding the full teardown sequence.	Low
General	Make resource close idempotent to prevent double-close The close() method is not idempotent — if called twice without claimOwnership(), it will attempt to close an already-closed VectorSchemaRoot, which can cause Arrow memory accounting errors or exceptions. A guard flag should be added to ensure the root is only closed once. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java [223-228] +private boolean closed = false; + /** Releases the consumer root unless {@link #claimOwnership()} was called. */ @Override public void close() { - if (!transferred && root != null) { + if (!transferred && !closed && root != null) { + closed = true; root.close(); } } Suggestion importance[1-10]: 5 __ Why: The double-close concern is valid for Arrow resources, but the improved_code places the closed field declaration outside the method body in an incorrect location. The logic is sound but the improved_code representation is not syntactically correct as presented.	Low
	Lazily initialize allocator to prevent resource leaks The allocator field is initialized inline at declaration time, which means it is allocated even if the plugin is never fully started or if an exception occurs during construction. If close() is never called (e.g., due to a test failure before plugin registration), the allocator leaks. Consider initializing it lazily in createComponents and handling null in close(). plugins/arrow-flight-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/NativeArrowTransportIT.java [504-533] public static class NativeArrowTestPlugin extends Plugin implements ActionPlugin { - private final BufferAllocator allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + private BufferAllocator allocator; ... @Override + public Collection<Object> createComponents(...) { + allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + return List.of(new TestAllocatorHolder(allocator)); + } + + @Override public void close() { - allocator.close(); + if (allocator != null) { + allocator.close(); + } } } Suggestion importance[1-10]: 4 __ Why: The concern about eager initialization causing leaks if close() is never called is valid, but in test infrastructure the risk is lower and the close() method already exists. The suggestion adds defensive null-check which is a minor improvement.	Low

Suggestions up to commit 7565fb9

Category	Suggestion	Impact
Possible issue	Guard against empty field vectors on send path If arrowResponse.getRoot().getFieldVectors() is empty, calling .get(0) will throw an IndexOutOfBoundsException. This is the same guard that was added in VectorStreamInput.forNativeArrow, and should be applied here as well to avoid a crash on the send path. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightOutboundHandler.java [162-165] +if (arrowResponse.getRoot().getFieldVectors().isEmpty()) { + throw new IllegalStateException("ArrowBatchResponse has no field vectors"); +} streamRoot = VectorSchemaRoot.create( arrowResponse.getRoot().getSchema(), arrowResponse.getRoot().getFieldVectors().get(0).getAllocator() ); Suggestion importance[1-10]: 7 __ Why: Calling .get(0) on an empty getFieldVectors() list will throw an IndexOutOfBoundsException. The same guard was added in VectorStreamInput.forNativeArrow, so consistency and correctness require it here too on the send path.	Medium
General	Prevent root allocator resource leak on shutdown The ROOT allocator is a static singleton that is never closed, which will cause resource leaks in tests and environments where the class is loaded multiple times. Consider adding a JVM shutdown hook or providing an explicit close() method to release the root allocator when the node shuts down. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java [32-34] private static final RootAllocator ROOT = AccessController.doPrivileged( - (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) + (PrivilegedAction<RootAllocator>) () -> { + RootAllocator root = new RootAllocator(Long.MAX_VALUE); + Runtime.getRuntime().addShutdownHook(new Thread(root::close, "arrow-root-allocator-shutdown")); + return root; + } ); Suggestion importance[1-10]: 5 __ Why: The static ROOT allocator is never closed, which could cause resource leaks. However, adding a JVM shutdown hook inside a static initializer is a debatable pattern and may not integrate well with OpenSearch's lifecycle management. The concern is valid but the proposed fix may not be the best approach.	Low
	Prevent allocator leak on plugin initialization failure The allocator field is initialized inline at field declaration time, before close() is available to clean it up if the constructor or plugin loading fails. If an exception is thrown during plugin initialization after this point, the allocator will be leaked. Consider initializing it lazily or inside a try-finally block in the constructor. plugins/arrow-flight-rpc/src/internalClusterTest/java/org/opensearch/arrow/flight/NativeArrowTransportIT.java [503-504] public static class NativeArrowTestPlugin extends Plugin implements ActionPlugin { - private final BufferAllocator allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + private final BufferAllocator allocator; + public NativeArrowTestPlugin() { + this.allocator = ArrowAllocatorProvider.newChildAllocator("native-arrow-test", Long.MAX_VALUE); + } + Suggestion importance[1-10]: 3 __ Why: Moving the allocator initialization to the constructor is a minor improvement for safety during initialization failures, but in practice plugin constructors rarely throw and this is test code, making the impact low.	Low

Suggestions up to commit a8e576b

Category	Suggestion	Impact
Possible issue	Guard against empty field vectors on send path If arrowResponse.getRoot().getFieldVectors() is empty, get(0) will throw an IndexOutOfBoundsException. This is the same guard that was added in VectorStreamInput.forNativeArrow, and should be applied here as well to avoid a crash on the send path. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightOutboundHandler.java [162-165] +if (arrowResponse.getRoot().getFieldVectors().isEmpty()) { + throw new IllegalStateException("ArrowBatchResponse has no field vectors"); +} streamRoot = VectorSchemaRoot.create( arrowResponse.getRoot().getSchema(), arrowResponse.getRoot().getFieldVectors().get(0).getAllocator() ); Suggestion importance[1-10]: 7 __ Why: The get(0) call on getFieldVectors() will throw an IndexOutOfBoundsException if the list is empty, which is a real crash risk on the send path. The same guard was already added in VectorStreamInput.forNativeArrow, making this a consistent and valid fix.	Medium
General	Prevent static root allocator resource leak The ROOT allocator is a static singleton that is never closed, which will cause resource leaks in tests and environments that reload classes. Consider adding a JVM shutdown hook or providing an explicit close() method to release the root allocator when the JVM exits. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java [32-34] private static final RootAllocator ROOT = AccessController.doPrivileged( - (PrivilegedAction<RootAllocator>) () -> new RootAllocator(Long.MAX_VALUE) + (PrivilegedAction<RootAllocator>) () -> { + RootAllocator root = new RootAllocator(Long.MAX_VALUE); + Runtime.getRuntime().addShutdownHook(new Thread(root::close, "arrow-root-allocator-shutdown")); + return root; + } ); Suggestion importance[1-10]: 4 __ Why: The static ROOT allocator is never closed, which could cause resource leaks. However, adding a JVM shutdown hook for a static singleton is a debatable pattern and may not be appropriate in all deployment scenarios. The suggestion is valid but has limited impact since the allocator is intentionally long-lived.	Low
	Verify child allocator closure ordering on shutdown rootAllocator is now a child of the shared ArrowAllocatorProvider root, but rootAllocator.getLimit() returns the child's own limit (Long.MAX_VALUE), which is correct. However, rootAllocator is a child allocator and calling rootAllocator.newChildAllocator(...) creates grandchildren of the shared root — this is fine, but the doStop/doClose logic that previously closed a RootAllocator must now close a child allocator. Verify that the existing doStop path closes rootAllocator (and thus its children) correctly, since closing a child allocator that still has open children will throw an IllegalStateException in Arrow. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java [146-148] rootAllocator = ArrowAllocatorProvider.newChildAllocator("flight", Long.MAX_VALUE); -serverAllocator = rootAllocator.newChildAllocator("server", 0, rootAllocator.getLimit()); -clientAllocator = rootAllocator.newChildAllocator("client", 0, rootAllocator.getLimit()); +serverAllocator = rootAllocator.newChildAllocator("server", 0, Long.MAX_VALUE); +clientAllocator = rootAllocator.newChildAllocator("client", 0, Long.MAX_VALUE); Suggestion importance[1-10]: 3 __ Why: The suggestion asks to verify shutdown behavior and changes rootAllocator.getLimit() to Long.MAX_VALUE directly, but the improved_code is functionally equivalent since rootAllocator was created with Long.MAX_VALUE. This is more of a verification concern than a concrete bug fix, and the code change itself is trivial.	Low
	Reset source row count after vector transfer After FlightUtils.transferRoot drains streamRoot into consumerRoot, the streamRoot's row count is not reset to zero. Depending on FlightUtils.transferRoot's implementation, the stream root may still report a non-zero row count, which could confuse the next flightStream.next() call or metrics. Confirm that transferRoot resets the source row count, or explicitly call streamRoot.setRowCount(0) after the transfer. plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java [64-79] static VectorStreamInput forNativeArrow(VectorSchemaRoot streamRoot, NamedWriteableRegistry registry) { if (streamRoot.getFieldVectors().isEmpty()) { throw new IllegalStateException("Native Arrow batch has no field vectors"); } VectorSchemaRoot consumerRoot = VectorSchemaRoot.create( streamRoot.getSchema(), streamRoot.getFieldVectors().getFirst().getAllocator() ); try { FlightUtils.transferRoot(streamRoot, consumerRoot); + streamRoot.setRowCount(0); } catch (Throwable t) { consumerRoot.close(); throw t; } return new NativeArrow(consumerRoot, registry); } Suggestion importance[1-10]: 3 __ Why: The suggestion to reset streamRoot's row count after transfer is a defensive measure, but the test in VectorStreamInputTests already asserts assertEquals("source must be drained", 0, shared.getRowCount()), suggesting FlightUtils.transferRoot already handles this. The suggestion is speculative and may be unnecessary.	Low

[github-actions]

Persistent review updated to latest commit 931a544

[github-actions]

❌ Gradle check result for 931a544: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit d3c2ffd

[github-actions]

✅ Gradle check result for d3c2ffd: SUCCESS

[codecov]

Codecov Report

❌ Patch coverage is 84.46602% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.33%. Comparing base (2b482e7) to head (6a09343).

Files with missing lines	Patch %	Lines
...arch/arrow/flight/transport/VectorStreamInput.java	92.18%	3 Missing and 2 partials ⚠️
...h/example/stream/StreamTransportExamplePlugin.java	0.00%	4 Missing ⚠️
.../arrow/flight/transport/FlightOutboundHandler.java	77.77%	1 Missing and 1 partial ⚠️
...rrow/flight/transport/FlightTransportResponse.java	71.42%	1 Missing and 1 partial ⚠️
...rch/arrow/flight/transport/ArrowBatchResponse.java	87.50%	0 Missing and 1 partial ⚠️
...e/stream/TransportNativeArrowStreamDataAction.java	0.00%	1 Missing ⚠️
...ing/handler/TraceableTransportResponseHandler.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #21454      +/-   ##
============================================
- Coverage     73.44%   73.33%   -0.11%     
+ Complexity    74456    74371      -85     
============================================
  Files          5967     5967              
  Lines        338232   338263      +31     
  Branches      48755    48761       +6     
============================================
- Hits         248399   248060     -339     
- Misses        70075    70402     +327     
- Partials      19758    19801      +43     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

PR Code Analyzer ❗

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

Path	Line	Severity	Description
plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/VectorStreamInput.java	131	medium	ensureCanReadBytes is implemented as a no-op, removing the inherited pre-read bounds-checking hook from StreamInput. While individual read methods in ByteSerialized do their own checking, silencing this safety hook could mask buffer-overread conditions in future subclasses that rely on the base class contract.
plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java	33	low	Static RootAllocator is initialized with Long.MAX_VALUE and is never closed. Combined with child allocators also created with Long.MAX_VALUE limits, this removes any programmatic upper bound on off-heap memory consumption, making the process susceptible to resource exhaustion if allocation paths are reachable by untrusted input.
plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/ArrowAllocatorProvider.java	30	low	Uses the deprecated and removal-scheduled AccessController.doPrivileged API suppressed with @SuppressWarnings("removal"). While common in legacy OpenSearch code, suppressing the warning hides that this API is being removed in newer JVM versions; code relying on it for security boundary enforcement will silently break.

The table above displays the top 10 most important findings.

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

---

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

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

---

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

Thanks.

[github-actions]

Persistent review updated to latest commit f6d9195

[github-actions]

Persistent review updated to latest commit 60898f5

[github-actions]

❌ Gradle check result for 60898f5: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 1313438

[github-actions]

Persistent review updated to latest commit afd7a6b

[github-actions]

Persistent review updated to latest commit 64446b8

[github-actions]

❌ Gradle check result for 64446b8: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 336e28b

[github-actions]

Persistent review updated to latest commit af2d6ee

[github-actions]

❕ Gradle check result for af2d6ee: UNSTABLE

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

[github-actions]

Persistent review updated to latest commit 6a09343

[github-actions]

❌ Gradle check result for 6a09343: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 6a09343

[github-actions]

❌ Gradle check result for 6a09343: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[github-actions]

Persistent review updated to latest commit 6a09343

[github-actions]

✅ Gradle check result for 6a09343: SUCCESS