Fix Flight integration tests using metadata

java/flight/src/main/java/org/apache/arrow/flight/ArrowMessage.java

@@ -307,7 +307,8 @@ class ArrowMessage implements AutoCloseable {
       cos.writeBytes(FlightData.DATA_HEADER_FIELD_NUMBER, bytes);
 
       if (appMetadata != null && appMetadata.capacity() > 0) {
-        cos.writeByteBuffer(FlightData.APP_METADATA_FIELD_NUMBER, appMetadata.asNettyBuffer().nioBuffer());
+        // Must call slice() as CodedOutputStream#writeByteBuffer writes -capacity- bytes, not -limit- bytes
+        cos.writeByteBuffer(FlightData.APP_METADATA_FIELD_NUMBER, appMetadata.asNettyBuffer().nioBuffer().slice());
         // This is weird, but implicitly, writing an ArrowMessage frees any references it has
         appMetadata.getReferenceManager().release();
       }

java/flight/src/main/java/org/apache/arrow/flight/example/integration/IntegrationTestClient.java

@@ -109,8 +109,9 @@ class IntegrationTestClient {
 
             @Override
             public void onNext(PutResult val) {
-              final String metadata = StandardCharsets.UTF_8.decode(val.getApplicationMetadata().nioBuffer())
-                  .toString();
+              final byte[] metadataRaw = new byte[val.getApplicationMetadata().readableBytes()];
+              val.getApplicationMetadata().readBytes(metadataRaw);
+              final String metadata = new String(metadataRaw, StandardCharsets.UTF_8);
               if (!Integer.toString(counter).equals(metadata)) {
                 throw new RuntimeException(
                     String.format("Invalid ACK from server. Expected '%d' but got '%s'.", counter, metadata));

python/pyarrow/tests/test_flight.py

@@ -28,7 +28,6 @@ import time
 import traceback
 
 import pytest
-
 import pyarrow as pa
 
 from pyarrow.compat import tobytes