Fix ClassCastException in FlightClientChannel for large requests

CHANGELOG.md

@@ -73,6 +73,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add S3Repository.LEGACY_MD5_CHECKSUM_CALCULATION to list of repository-s3 settings ([#19788](https://github.com/opensearch-project/OpenSearch/pull/19788))
 - Fix NullPointerException when restoring remote snapshot with missing shard size information ([#19684](https://github.com/opensearch-project/OpenSearch/pull/19684))
 - Fix NPE of ScriptScoreQuery ([#19650](https://github.com/opensearch-project/OpenSearch/pull/19650))
+- Fix ClassCastException in FlightClientChannel for requests larger than 16KB ([#20010](https://github.com/opensearch-project/OpenSearch/pull/20010))
 
 ### Dependencies
 - Update to Gradle 9.2 ([#19575](https://github.com/opensearch-project/OpenSearch/pull/19575)) ([#19856](https://github.com/opensearch-project/OpenSearch/pull/19856))

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

@@ -18,7 +18,6 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.core.action.ActionListener;
-import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.common.transport.BoundTransportAddress;
@@ -35,7 +34,6 @@
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -351,8 +349,7 @@ private void notifyListener(ActionListener<Void> listener, CompletableFuture<Voi
     }
 
     private Ticket serializeToTicket(BytesReference reference) {
-        byte[] data = Arrays.copyOfRange(((BytesArray) reference).array(), 0, reference.length());
-        return new Ticket(data);
+        return new Ticket(BytesReference.toBytes(reference));
     }
 
     @Override

plugins/arrow-flight-rpc/src/test/java/org/opensearch/arrow/flight/transport/FlightClientChannelTests.java

@@ -13,6 +13,7 @@
 import org.opensearch.core.common.bytes.BytesArray;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.StreamInput;
+import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.transport.TransportResponse;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.StreamTransportResponseHandler;
@@ -588,4 +589,82 @@ public void testSetMessageListenerTwice() {
         IllegalStateException exception = assertThrows(IllegalStateException.class, () -> flightTransport.setMessageListener(listener2));
         assertEquals("Cannot set message listener twice", exception.getMessage());
     }
+
+    static class LargeTestRequest extends TestRequest {
+        private final String largeData;
+
+        LargeTestRequest(String data) {
+            this.largeData = data;
+        }
+
+        LargeTestRequest(StreamInput in) throws IOException {
+            super(in);
+            this.largeData = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(largeData);
+        }
+    }
+
+    public void testLargeRequest() throws Exception {
+        String action = "internal:test/large";
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Exception> error = new AtomicReference<>();
+
+        streamTransportService.registerRequestHandler(action, ThreadPool.Names.SAME, LargeTestRequest::new, (request, channel, task) -> {
+            try {
+                channel.sendResponseBatch(new TestResponse("OK"));
+                channel.completeStream();
+            } catch (Exception e) {
+                try {
+                    channel.sendResponse(e);
+                } catch (IOException ex) {}
+            }
+        });
+
+        LargeTestRequest testRequest = new LargeTestRequest("X".repeat(20 * 1024));
+
+        streamTransportService.sendRequest(
+            remoteNode,
+            action,
+            testRequest,
+            TransportRequestOptions.builder().withType(TransportRequestOptions.Type.STREAM).build(),
+            new StreamTransportResponseHandler<TestResponse>() {
+                @Override
+                public void handleStreamResponse(StreamTransportResponse<TestResponse> streamResponse) {
+                    try {
+                        while (streamResponse.nextResponse() != null) {
+                        }
+                        streamResponse.close();
+                    } catch (Exception e) {
+                        error.set(e);
+                    } finally {
+                        latch.countDown();
+                    }
+                }
+
+                @Override
+                public void handleException(TransportException exp) {
+                    error.set(exp);
+                    latch.countDown();
+                }
+
+                @Override
+                public String executor() {
+                    return ThreadPool.Names.SAME;
+                }
+
+                @Override
+                public TestResponse read(StreamInput in) throws IOException {
+                    return new TestResponse(in);
+                }
+            }
+        );
+
+        assertTrue(latch.await(TIMEOUT_SEC, TimeUnit.SECONDS));
+        assertNull(error.get());
+    }
 }