apacheGH-38022: [Java][FlightRPC] Expose app_metadata on FlightInfo a…

…nd FlightEndpoint (apache#38331)

Making necessary changes in Java to expose the newly added app_metadata.
* Closes: apache#38022

Authored-by: Diego Fernandez <aiguo.fernandez@gmail.com>
Signed-off-by: David Li <li.davidm96@gmail.com>

dev/archery/archery/integration/runner.py

@@ -615,7 +615,7 @@ def run_all_tests(with_cpp=True, with_java=True, with_js=True,
  Scenario(
  "app_metadata_flight_info_endpoint",
  description="Ensure support FlightInfo and Endpoint app_metadata",
- skip_testers={"JS", "C#", "Rust", "Java"}
+ skip_testers={"JS", "C#", "Rust"}
  ),
  Scenario(
  "flight_sql",

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

@@ -23,13 +23,15 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 
 import org.apache.arrow.flight.impl.Flight;
 
+import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
 
 /**
@@ -39,6 +41,7 @@ public class FlightEndpoint {
  private final List<Location> locations;
  private final Ticket ticket;
  private final Instant expirationTime;
+ private final byte[] appMetadata;
 
  /**
  * Constructs a new endpoint with no expiration time.
@@ -54,13 +57,22 @@ public FlightEndpoint(Ticket ticket, Location... locations) {
  * Constructs a new endpoint with an expiration time.
  *
  * @param ticket A ticket that describe the key of a data stream.
+ * @param expirationTime (optional) When this endpoint expires.
  * @param locations The possible locations the stream can be retrieved from.
  */
  public FlightEndpoint(Ticket ticket, Instant expirationTime, Location... locations) {
+ this(ticket, expirationTime, null, Collections.unmodifiableList(new ArrayList<>(Arrays.asList(locations))));
+ }
+
+ /**
+ * Private constructor with all parameters. Should only be called by Builder.
+ */
+ private FlightEndpoint(Ticket ticket, Instant expirationTime, byte[] appMetadata, List<Location> locations) {
  Objects.requireNonNull(ticket);
- this.locations = Collections.unmodifiableList(new ArrayList<>(Arrays.asList(locations)));
+ this.locations = locations;
  this.expirationTime = expirationTime;
  this.ticket = ticket;
+ this.appMetadata = appMetadata;
  }
 
  /**
@@ -77,6 +89,7 @@ public FlightEndpoint(Ticket ticket, Instant expirationTime, Location... locatio
  } else {
  this.expirationTime = null;
  }
+ this.appMetadata = (flt.getAppMetadata().size() == 0 ? null : flt.getAppMetadata().toByteArray());
  this.ticket = new Ticket(flt.getTicket());
  }
 
@@ -92,6 +105,10 @@ public Optional<Instant> getExpirationTime() {
  return Optional.ofNullable(expirationTime);
  }
 
+ public byte[] getAppMetadata() {
+ return appMetadata;
+ }
+
  /**
  * Converts to the protocol buffer representation.
  */
@@ -111,6 +128,10 @@ Flight.FlightEndpoint toProtocol() {
  .build());
  }
 
+ if (appMetadata != null) {
+ b.setAppMetadata(ByteString.copyFrom(appMetadata));
+ }
+
  return b.build();
  }
 
@@ -148,12 +169,13 @@ public boolean equals(Object o) {
  FlightEndpoint that = (FlightEndpoint) o;
  return locations.equals(that.locations) &&
  ticket.equals(that.ticket) &&
- Objects.equals(expirationTime, that.expirationTime);
+ Objects.equals(expirationTime, that.expirationTime) &&
+ Arrays.equals(appMetadata, that.appMetadata);
  }
 
  @Override
  public int hashCode() {
- return Objects.hash(locations, ticket, expirationTime);
+ return Objects.hash(locations, ticket, expirationTime, Arrays.hashCode(appMetadata));
  }
 
  @Override
@@ -162,6 +184,59 @@ public String toString() {
  "locations=" + locations +
  ", ticket=" + ticket +
  ", expirationTime=" + (expirationTime == null ? "(none)" : expirationTime.toString()) +
+ ", appMetadata=" + (appMetadata == null ? "(none)" : Base64.getEncoder().encodeToString(appMetadata)) +
  '}';
  }
+
+ /**
+ * Create a builder for FlightEndpoint.
+ *
+ * @param ticket A ticket that describe the key of a data stream.
+ * @param locations The possible locations the stream can be retrieved from.
+ */
+ public static Builder builder(Ticket ticket, Location... locations) {
+ return new Builder(ticket, locations);
+ }
+
+ /**
+ * Builder for FlightEndpoint.
+ */
+ public static final class Builder {
+ private final Ticket ticket;
+ private final List<Location> locations;
+ private Instant expirationTime = null;
+ private byte[] appMetadata = null;
+
+ private Builder(Ticket ticket, Location... locations) {
+ this.ticket = ticket;
+ this.locations = Collections.unmodifiableList(new ArrayList<>(Arrays.asList(locations)));
+ }
+
+ /**
+ * Set expiration time for the endpoint. Default is null, which means don't expire.
+ *
+ * @param expirationTime (optional) When this endpoint expires.
+ */
+ public Builder setExpirationTime(Instant expirationTime) {
+ this.expirationTime = expirationTime;
+ return this;
+ }
+
+ /**
+ * Set the app metadata to send along with the flight. Default is null;
+ *
+ * @param appMetadata Metadata to send along with the flight
+ */
+ public Builder setAppMetadata(byte[] appMetadata) {
+ this.appMetadata = appMetadata;
+ return this;
+ }
+
+ /**
+ * Build FlightEndpoint object.
+ */
+ public FlightEndpoint build() {
+ return new FlightEndpoint(ticket, expirationTime, appMetadata, locations);
+ }
+ }
 }

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

@@ -23,6 +23,8 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -51,6 +53,7 @@ public class FlightInfo {
  private final long records;
  private final boolean ordered;
  private final IpcOption option;
+ private final byte[] appMetadata;
 
  /**
  * Constructs a new instance.
@@ -94,6 +97,23 @@ public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoin
  */
  public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoint> endpoints, long bytes,
  long records, boolean ordered, IpcOption option) {
+ this(schema, descriptor, endpoints, bytes, records, ordered, option, null);
+ }
+
+ /**
+ * Constructs a new instance.
+ *
+ * @param schema The schema of the Flight
+ * @param descriptor An identifier for the Flight.
+ * @param endpoints A list of endpoints that have the flight available.
+ * @param bytes The number of bytes in the flight
+ * @param records The number of records in the flight.
+ * @param ordered Whether the endpoints in this flight are ordered.
+ * @param option IPC write options.
+ * @param appMetadata Metadata to send along with the flight
+ */
+ public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoint> endpoints, long bytes,
+ long records, boolean ordered, IpcOption option, byte[] appMetadata) {
  Objects.requireNonNull(descriptor);
  Objects.requireNonNull(endpoints);
  if (schema != null) {
@@ -106,6 +126,7 @@ public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoin
  this.records = records;
  this.ordered = ordered;
  this.option = option;
+ this.appMetadata = appMetadata;
  }
 
  /**
@@ -131,6 +152,7 @@ public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoin
  bytes = pbFlightInfo.getTotalBytes();
  records = pbFlightInfo.getTotalRecords();
  ordered = pbFlightInfo.getOrdered();
+ appMetadata = (pbFlightInfo.getAppMetadata().size() == 0 ? null : pbFlightInfo.getAppMetadata().toByteArray());
  option = IpcOption.DEFAULT;
  }
 
@@ -167,6 +189,10 @@ public boolean getOrdered() {
  return ordered;
  }
 
+ public byte[] getAppMetadata() {
+ return appMetadata;
+ }
+
  /**
  * Converts to the protocol buffer representation.
  */
@@ -189,6 +215,9 @@ Flight.FlightInfo toProtocol() {
  throw new RuntimeException(e);
  }
  }
+ if (appMetadata != null) {
+ builder.setAppMetadata(ByteString.copyFrom(appMetadata));
+ }
  return builder.build();
  }
 
@@ -229,12 +258,13 @@ public boolean equals(Object o) {
  schema.equals(that.schema) &&
  descriptor.equals(that.descriptor) &&
  endpoints.equals(that.endpoints) &&
- ordered == that.ordered;
+ ordered == that.ordered &&
+ Arrays.equals(appMetadata, that.appMetadata);
  }
 
  @Override
  public int hashCode() {
- return Objects.hash(schema, descriptor, endpoints, bytes, records, ordered);
+ return Objects.hash(schema, descriptor, endpoints, bytes, records, ordered, Arrays.hashCode(appMetadata));
  }
 
  @Override
@@ -246,6 +276,95 @@ public String toString() {
  ", bytes=" + bytes +
  ", records=" + records +
  ", ordered=" + ordered +
+ ", appMetadata=" + (appMetadata == null ? "(none)" : Base64.getEncoder().encodeToString(appMetadata)) +
  '}';
  }
+
+ /**
+ * Create a builder for FlightInfo.
+ *
+ * @param schema The schema of the Flight
+ * @param descriptor An identifier for the Flight.
+ * @param endpoints A list of endpoints that have the flight available.
+ */
+ public static Builder builder(Schema schema, FlightDescriptor descriptor, List<FlightEndpoint> endpoints) {
+ return new Builder(schema, descriptor, endpoints);
+ }
+
+ /**
+ * Builder for FlightInfo.
+ */
+ public static final class Builder {
+ private final Schema schema;
+ private final FlightDescriptor descriptor;
+ private final List<FlightEndpoint> endpoints;
+ private long bytes = -1;
+ private long records = -1;
+ private boolean ordered = false;
+ private IpcOption option = IpcOption.DEFAULT;
+ private byte[] appMetadata = null;
+
+ private Builder(Schema schema, FlightDescriptor descriptor, List<FlightEndpoint> endpoints) {
+ this.schema = schema;
+ this.descriptor = descriptor;
+ this.endpoints = endpoints;
+ }
+
+ /**
+ * Set the number of bytes for the flight. Default to -1 for unknown.
+ *
+ * @param bytes The number of bytes in the flight
+ */
+ public Builder setBytes(long bytes) {
+ this.bytes = bytes;
+ return this;
+ }
+
+ /**
+ * Set the number of records for the flight. Default to -1 for unknown.
+ *
+ * @param records The number of records in the flight.
+ */
+ public Builder setRecords(long records) {
+ this.records = records;
+ return this;
+ }
+
+ /**
+ * Set whether the flight endpoints are ordered. Default is false.
+ *
+ * @param ordered Whether the endpoints in this flight are ordered.
+ */
+ public Builder setOrdered(boolean ordered) {
+ this.ordered = ordered;
+ return this;
+ }
+
+ /**
+ * Set IPC write options. Default is IpcOption.DEFAULT
+ *
+ * @param option IPC write options.
+ */
+ public Builder setOption(IpcOption option) {
+ this.option = option;
+ return this;
+ }
+
+ /**
+ * Set the app metadata to send along with the flight. Default is null.
+ *
+ * @param appMetadata Metadata to send along with the flight
+ */
+ public Builder setAppMetadata(byte[] appMetadata) {
+ this.appMetadata = appMetadata;
+ return this;
+ }
+
+ /**
+ * Build FlightInfo object.
+ */
+ public FlightInfo build() {
+ return new FlightInfo(schema, descriptor, endpoints, bytes, records, ordered, option, appMetadata);
+ }
+ }
 }

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

@@ -113,10 +113,13 @@ public void roundTripInfo() throws Exception {
  Field.nullable("a", new ArrowType.Int(32, true)),
  Field.nullable("b", new ArrowType.FixedSizeBinary(32))
  ), metadata);
- final FlightInfo info1 = new FlightInfo(schema, FlightDescriptor.path(), Collections.emptyList(), -1, -1);
+ final FlightInfo info1 = FlightInfo.builder(schema, FlightDescriptor.path(), Collections.emptyList())
+ .setAppMetadata("foo".getBytes()).build();
  final FlightInfo info2 = new FlightInfo(schema, FlightDescriptor.command(new byte[2]),
- Collections.singletonList(new FlightEndpoint(
- new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock"))), 200, 500);
+ Collections.singletonList(
+ FlightEndpoint.builder(new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock"))
+ .setAppMetadata("bar".getBytes()).build()
+ ), 200, 500);
  final FlightInfo info3 = new FlightInfo(schema, FlightDescriptor.path("a", "b"),
  Arrays.asList(new FlightEndpoint(
  new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock")),

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

@@ -28,6 +28,7 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.vector.ipc.message.IpcOption;
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
@@ -137,7 +138,8 @@ public void supportsNullSchemas() throws Exception
  @Override
  public FlightInfo getFlightInfo(CallContext context,
  FlightDescriptor descriptor) {
- return new FlightInfo(null, descriptor, Collections.emptyList(), 0, 0);
+ return new FlightInfo(null, descriptor, Collections.emptyList(),
+ 0, 0, false, IpcOption.DEFAULT, "foo".getBytes());
  }
  };
 
@@ -147,6 +149,7 @@ public FlightInfo getFlightInfo(CallContext context,
  FlightInfo flightInfo = client.getInfo(FlightDescriptor.path("test"));
  Assertions.assertEquals(Optional.empty(), flightInfo.getSchemaOptional());
  Assertions.assertEquals(new Schema(Collections.emptyList()), flightInfo.getSchema());
+ Assertions.assertArrayEquals(flightInfo.getAppMetadata(), "foo".getBytes());
 
  Exception e = Assertions.assertThrows(
  FlightRuntimeException.class,