apacheGH-35888: [Java] Add FlightStatusCode.RESOURCE_EXHAUSTED (apach…

…e#41508)

### Rationale for this change

Related to apache#35888

Currently the gRPC Status.RESOURCE_EXHAUSTED exception/code is translated by the Java FlightServer into FlightStatusCode.INVALID_ARGUMENT and thrown to the client as gRPC INVALID_ARGUMENT exception.

That may mislead the other party as the INVALID_ARGUMENT indicates an input parameters problem where in reality the backed server intention was rather 'back off and try later'.

### What changes are included in this PR?

Add the FlightStatusCode.RESOURCE_EXHAUSTED code and make sure is translated from/to the gRPC Status.RESOURCE_EXHAUSTED

### Are these changes tested?

Unit tests included to validate the RESOURCE_EXHAUSTED translation  between flight and grpc codes.

### Are there any user-facing changes?

No.

Users may start seeing RESOURCE_EXHAUSTED instead of INVALID_ARGUMENT code. In both cases this is an exception seen on the client side so I am considering this as a _not breaking change to any public API_.

Although, may have an influence in the client side flows if one decided to react conditionally on exception status code.

* GitHub Issue: apache#35888

Authored-by: Jacek Stania <janosik47@gmail.com>
Signed-off-by: David Li <li.davidm96@gmail.com>

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

@@ -49,6 +49,7 @@ public class CallStatus {
   public static final CallStatus UNAUTHORIZED = FlightStatusCode.UNAUTHORIZED.toStatus();
   public static final CallStatus UNIMPLEMENTED = FlightStatusCode.UNIMPLEMENTED.toStatus();
   public static final CallStatus UNAVAILABLE = FlightStatusCode.UNAVAILABLE.toStatus();
+  public static final CallStatus RESOURCE_EXHAUSTED = FlightStatusCode.RESOURCE_EXHAUSTED.toStatus();
 
   /**
    * Create a new status.

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

@@ -71,6 +71,11 @@ public enum FlightStatusCode {
    * should send this code only if it has not done any work.
    */
   UNAVAILABLE,
+  /**
+   * Some resource has been exhausted, perhaps a per-user quota, or perhaps the entire file system is out of space.
+   * (see: https://grpc.github.io/grpc/core/md_doc_statuscodes.html)
+   */
+  RESOURCE_EXHAUSTED
   ;
 
   /**

java/flight/flight-core/src/main/java/org/apache/arrow/flight/grpc/StatusUtils.java

@@ -74,6 +74,8 @@ public static Status.Code toGrpcStatusCode(FlightStatusCode code) {
         return Code.UNIMPLEMENTED;
       case UNAVAILABLE:
         return Code.UNAVAILABLE;
+      case RESOURCE_EXHAUSTED:
+        return Code.RESOURCE_EXHAUSTED;
       default:
         return Code.UNKNOWN;
     }
@@ -101,7 +103,7 @@ public static FlightStatusCode fromGrpcStatusCode(Status.Code code) {
       case PERMISSION_DENIED:
         return FlightStatusCode.UNAUTHORIZED;
       case RESOURCE_EXHAUSTED:
-        return FlightStatusCode.INVALID_ARGUMENT;
+        return FlightStatusCode.RESOURCE_EXHAUSTED;
       case FAILED_PRECONDITION:
         return FlightStatusCode.INVALID_ARGUMENT;
       case ABORTED:

java/flight/flight-core/src/test/java/org/apache/arrow/flight/grpc/TestStatusUtils.java

@@ -48,4 +48,26 @@ public void testParseTrailers() {
     Assertions.assertTrue(callStatus.metadata().containsKey("content-type"));
     Assertions.assertEquals("text/html", callStatus.metadata().get("content-type"));
   }
+
+  @Test
+  public void testGrpcResourceExhaustedTranslatedToFlightStatus() {
+    Status status = Status.RESOURCE_EXHAUSTED;
+
+    CallStatus callStatus = StatusUtils.fromGrpcStatus(status);
+    Assertions.assertEquals(FlightStatusCode.RESOURCE_EXHAUSTED, callStatus.code());
+
+    FlightStatusCode flightStatusCode = StatusUtils.fromGrpcStatusCode(status.getCode());
+    Assertions.assertEquals(FlightStatusCode.RESOURCE_EXHAUSTED, flightStatusCode);
+  }
+
+  @Test
+  public void testFlightResourceExhaustedTranslatedToGrpcStatua() {
+    CallStatus callStatus = CallStatus.RESOURCE_EXHAUSTED;
+
+    Status.Code grpcStatusCode = StatusUtils.toGrpcStatusCode(callStatus.code());
+    Assertions.assertEquals(Status.RESOURCE_EXHAUSTED.getCode(), grpcStatusCode);
+
+    Status grpcStatus = StatusUtils.toGrpcStatus(callStatus);
+    Assertions.assertEquals(Status.RESOURCE_EXHAUSTED.getCode(), grpcStatus.getCode());
+  }
 }