remove redundant checks. add test to check fallback works

Author: xborder

flight/flight-core/src/main/java/org/apache/arrow/flight/FlightDataParser.java

@@ -24,6 +24,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
+import java.util.Objects;
 import org.apache.arrow.flight.impl.Flight.FlightData;
 import org.apache.arrow.flight.impl.Flight.FlightDescriptor;
 import org.apache.arrow.memory.ArrowBuf;
@@ -274,19 +275,10 @@ static ArrowBufReader tryArrowBufReader(BufferAllocator allocator, InputStream s
       }
 
       InputStream detachedStream = ((Detachable) stream).detach();
-      if (!(detachedStream instanceof HasByteBuffer)) {
-        closeQuietly(detachedStream);
-        return null;
-      }
-
       ByteBuffer detachedBuffer = ((HasByteBuffer) detachedStream).getByteBuffer();
-      if (detachedBuffer == null || !detachedBuffer.isDirect()) {
-        closeQuietly(detachedStream);
-        return null;
-      }
 
       long bufferAddress = MemoryUtil.getByteBufferAddress(detachedBuffer);
-      int bufferSize = detachedBuffer.remaining();
+      int bufferSize = Objects.requireNonNull(detachedBuffer).remaining();
 
       ForeignAllocation foreignAllocation =
           new ForeignAllocation(bufferSize, bufferAddress + detachedBuffer.position()) {

flight/flight-core/src/test/java/org/apache/arrow/flight/TestArrowMessageParse.java

@@ -275,6 +275,39 @@ public void testHeapBufferFallbackDoesNotDetach() throws Exception {
     }
   }
 
+  /** Verifies fallback to InputStream path when getByteBuffer() returns null. */
+  @Test
+  public void testNullByteBufferFallbackToInputStream() throws Exception {
+    byte[] appMetadataBytes = new byte[] {20, 21, 22};
+    byte[] bodyBytes = new byte[] {30, 31, 32, 33};
+    FlightDescriptor expectedDescriptor = createTestDescriptor();
+
+    byte[] serialized = buildFlightDataWithBothFields(appMetadataBytes, bodyBytes);
+    MockGrpcInputStream stream =
+        new MockGrpcInputStream(
+            ByteBuffer.wrap(serialized),
+            false);
+
+    try (ArrowMessage message = ArrowMessage.createMarshaller(allocator).parse(stream)) {
+      assertEquals(expectedDescriptor, message.getDescriptor());
+
+      ArrowBuf appMetadata = message.getApplicationMetadata();
+      assertNotNull(appMetadata);
+      byte[] actualAppMetadata = new byte[appMetadataBytes.length];
+      appMetadata.getBytes(0, actualAppMetadata);
+      assertArrayEquals(appMetadataBytes, actualAppMetadata);
+
+      ArrowBuf body = Iterables.getOnlyElement(message.getBufs());
+      assertNotNull(body);
+      byte[] actualBody = new byte[bodyBytes.length];
+      body.getBytes(0, actualBody);
+      assertArrayEquals(bodyBytes, actualBody);
+
+      assertEquals(0, stream.getDetachCount());
+    }
+    assertEquals(0, allocator.getAllocatedMemory());
+  }
+
   // Helper methods to build complete FlightData messages
 
   private FlightDescriptor createTestDescriptor() {