Cleanup pass on the singleton-removal migration

Self-review surfaced six items worth fixing before pushing:

* DefaultPlanExecutor now subscribes to the QUERY pool listener and updates coordinatorAllocator on dynamic resize, matching the pattern used by AnalyticsSearchService and FlightTransport. Without this, runtime grows of parquet.native.pool.query.max never reached the coordinator-side allocator.
* DefaultPlanExecutor.coordinatorAllocator is now sized at queryPool.getLimit() instead of Long.MAX_VALUE so the literal matches the actual cap.
* Drop the dead try/catch around removeListener in FlightTransport.stopInternal and AnalyticsSearchService.close. Those catches were holdovers from the singleton era and can't fire now that nativeAllocator is constructor-injected.
* Fix the joined-import line in AnalyticsSearchService that two earlier patches concatenated.
* Expand ArrowBatchResponse Javadoc with a pool-selection decision tree (FLIGHT for transport, INGEST for ingest path, QUERY for query execution) so readers know which pool to pick.
* Update misleading Javadoc in TransportNativeArrowStreamDataAction and DefaultPlanExecutor that referenced non-existent close paths. The pre-existing leak is preserved (out of scope for this PR) but documented honestly with a path forward.

Signed-off-by: Gaurav Singh <snghsvn@amazon.com>

Author: gaurav-amz

plugins/arrow-base/src/main/java/org/opensearch/arrow/allocator/NativeAllocatorPluginStats.java

@@ -30,12 +30,21 @@ public class NativeAllocatorPluginStats implements PluginNodeStats {
 
     private final NativeAllocatorPoolStats poolStats;
 
-    /** Wraps the given snapshot. */
+    /**
+     * Wraps the given snapshot.
+     *
+     * @param poolStats the pool stats snapshot to expose under {@code _nodes/stats}
+     */
     public NativeAllocatorPluginStats(NativeAllocatorPoolStats poolStats) {
         this.poolStats = poolStats;
     }
 
-    /** Reads from the wire. */
+    /**
+     * Reads from the wire.
+     *
+     * @param in the stream input
+     * @throws IOException if reading fails
+     */
     public NativeAllocatorPluginStats(StreamInput in) throws IOException {
         this.poolStats = new NativeAllocatorPoolStats(in);
     }

plugins/arrow-base/src/main/java/org/opensearch/arrow/transport/ArrowBatchResponse.java

@@ -48,9 +48,17 @@
  * <p><b>Allocator rules:</b>
  * <ul>
  *   <li><b>Send side:</b> Source the allocator from one of the framework's named pools via
- *       {@link ArrowNativeAllocator#getPoolAllocator(String)} (e.g. {@code POOL_FLIGHT},
- *       {@code POOL_INGEST}, {@code POOL_QUERY}). All allocators must share the same root so
- *       zero-copy transfers pass Arrow's {@code AllocationManager} associate check.</li>
+ *       {@link ArrowNativeAllocator#getPoolAllocator(String)}. Pick the pool that matches the
+ *       semantics of the producing component:
+ *       <ul>
+ *         <li>{@code POOL_FLIGHT} — transport-layer producers/consumers (arrow-flight-rpc and
+ *             plugins built on top of {@code StreamTransportService}).</li>
+ *         <li>{@code POOL_INGEST} — ingest-path producers (parquet-data-format VSR allocators).</li>
+ *         <li>{@code POOL_QUERY} — query-execution producers (analytics-engine fragments and
+ *             coordinator-side intermediate batches).</li>
+ *       </ul>
+ *       All allocators must share the same root so zero-copy transfers pass Arrow's
+ *       {@code AllocationManager} associate check.</li>
  *   <li><b>Send side:</b> Allocators must outlive the transport stream — some transports
  *       (e.g., gRPC zero-copy) retain buffer references beyond stream completion. Do not
  *       create and close a child allocator per request.</li>

plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/transport/FlightTransport.java

@@ -282,11 +282,7 @@ private List<InetSocketAddress> bindToPort(InetAddress[] hostAddresses) {
     protected void stopInternal() {
         try {
             if (poolListener != null) {
-                try {
-                    nativeAllocator.removeListener(poolListener);
-                } catch (IllegalStateException ignored) {
-                    // Framework already torn down — nothing to detach from.
-                }
+                nativeAllocator.removeListener(poolListener);
                 poolListener = null;
             }
             if (flightServer != null) {

plugins/examples/stream-transport-example/src/main/java/org/opensearch/example/stream/TransportNativeArrowStreamDataAction.java

@@ -38,12 +38,18 @@
  *
  * <p>Demonstrates the pipelined producer pattern:
  * <ol>
- *   <li>Receive an allocator owned by the plugin (closed in {@link StreamTransportExamplePlugin#close()})</li>
+ *   <li>Receive an allocator sourced from the framework's FLIGHT pool</li>
  *   <li>For each batch, create a {@link VectorSchemaRoot}, populate it, and wrap it in a response</li>
  *   <li>Send via {@code sendResponseBatch()} — the framework zero-copy transfers
  *       the vectors into the Flight stream on the executor thread</li>
  *   <li>Call {@code completeStream()} when done</li>
  * </ol>
+ *
+ * <p><b>Known leak:</b> the action's allocator is not closed on plugin teardown
+ * (the action is Guice-managed and not held by {@link StreamTransportExamplePlugin}).
+ * On node shutdown, ArrowNativeAllocator.close() will warn about this outstanding child.
+ * A clean fix would make this action {@link java.io.Closeable} and have the plugin track
+ * and close it from {@code Plugin#close()}.
  */
 public class TransportNativeArrowStreamDataAction extends TransportAction<NativeArrowStreamDataRequest, NativeArrowStreamDataResponse> {
 

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/AnalyticsSearchService.java

@@ -25,7 +25,8 @@
 import org.opensearch.analytics.spi.InstructionNode;
 import org.opensearch.arrow.allocator.ArrowNativeAllocator;
 import org.opensearch.arrow.spi.NativeAllocatorListener;
-import org.opensearch.arrow.spi.NativeAllocatorPoolConfig;import org.opensearch.common.concurrent.GatedCloseable;
+import org.opensearch.arrow.spi.NativeAllocatorPoolConfig;
+import org.opensearch.common.concurrent.GatedCloseable;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.tasks.TaskCancelledException;
 import org.opensearch.index.engine.exec.IndexReaderProvider;
@@ -113,12 +114,7 @@ public AnalyticsSearchService(
 
     @Override
     public void close() {
-        try {
-            nativeAllocator.removeListener(poolListener);
-        } catch (Exception ignored) {
-            // Best-effort — framework may have already torn down ahead of us in some
-            // shutdown paths.
-        }
+        nativeAllocator.removeListener(poolListener);
         allocator.close();
     }
 

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/DefaultPlanExecutor.java

@@ -33,6 +33,7 @@
 import org.opensearch.analytics.planner.dag.PlanForker;
 import org.opensearch.analytics.planner.dag.QueryDAG;
 import org.opensearch.arrow.allocator.ArrowNativeAllocator;
+import org.opensearch.arrow.spi.NativeAllocatorListener;
 import org.opensearch.arrow.spi.NativeAllocatorPoolConfig;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
@@ -76,9 +77,15 @@ public class DefaultPlanExecutor extends HandledTransportAction<ActionRequest, A
     private final Executor searchExecutor;
     private final TaskManager taskManager;
     private final NodeClient client;
-    // TODO: close on shutdown — currently arrow-base's root.close() will warn about this
-    // outstanding child. Consider wrapping in a Guice-bound type owned by AnalyticsPlugin.
+    private final ArrowNativeAllocator nativeAllocator;
+    // Pre-existing leak: DefaultPlanExecutor is Guice-instantiated and not closed by the
+    // framework, so coordinatorAllocator + the registered NativeAllocatorListener leak on
+    // node shutdown. ArrowNativeAllocator.close() will warn about the outstanding child.
+    // Fix would require either (a) making DefaultPlanExecutor LifecycleComponent so Node
+    // closes it, or (b) wrapping in a Guice-bound singleton that AnalyticsPlugin owns and
+    // closes from its own close() method. Out of scope for the SPI-removal migration.
     private final BufferAllocator coordinatorAllocator;
+    private final NativeAllocatorListener poolListener;
     private volatile long perQueryBufferLimit;
 
     @Inject
@@ -102,16 +109,28 @@ public DefaultPlanExecutor(
         this.taskManager = transportService.getTaskManager();
         this.client = client;
         this.scheduler = scheduler;
+        this.nativeAllocator = nativeAllocator;
         // Source the coordinator allocator from the framework's QUERY pool so coordinator-side
         // allocations (held shard responses, intermediate batches during reduce) count against
         // parquet.native.pool.query.max alongside the data-node-side per-fragment allocators in
         // AnalyticsSearchService. Before this, coordinator allocations were root-siblings outside
         // any pool and invisible to _nodes/stats.native_allocator.pools.query.allocated.
-        this.coordinatorAllocator = nativeAllocator.getPoolAllocator(NativeAllocatorPoolConfig.POOL_QUERY)
-            .newChildAllocator("coordinator", 0, Long.MAX_VALUE);
+        BufferAllocator queryPool = nativeAllocator.getPoolAllocator(NativeAllocatorPoolConfig.POOL_QUERY);
+        this.coordinatorAllocator = queryPool.newChildAllocator("coordinator", 0, queryPool.getLimit());
         this.perQueryBufferLimit = AnalyticsPlugin.COORDINATOR_BUFFER_LIMIT.get(clusterService.getSettings());
         clusterService.getClusterSettings()
             .addSettingsUpdateConsumer(AnalyticsPlugin.COORDINATOR_BUFFER_LIMIT, v -> perQueryBufferLimit = v);
+
+        // Subscribe to QUERY pool resize events so a dynamic update to
+        // parquet.native.pool.query.max propagates into this captured child allocator,
+        // matching the listener pattern used by AnalyticsSearchService and FlightTransport.
+        // Without this, runtime grows are silently ignored once a query is in flight.
+        this.poolListener = (poolName, newLimit) -> {
+            if (NativeAllocatorPoolConfig.POOL_QUERY.equals(poolName)) {
+                coordinatorAllocator.setLimit(newLimit);
+            }
+        };
+        nativeAllocator.addListener(poolListener);
     }
 
     @Override