[fix](arrow-flight) Let GetFlightInfo pass the session layer's Flight status through again

fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java

@@ -24,7 +24,6 @@
 import org.apache.doris.arrowflight.results.FlightSqlEndpointsLocation;
 import org.apache.doris.arrowflight.results.FlightSqlResultCacheEntry;
 import org.apache.doris.arrowflight.sessions.FlightSessionsManager;
-import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.IncrWindowNotReadyException;
 import org.apache.doris.common.Status;
 import org.apache.doris.common.util.DebugUtil;
@@ -351,18 +350,14 @@ public FlightInfo getFlightInfoStatement(final CommandStatementQuery request, fi
             return FlightProtocolAdapter.of(connectContext).callCommand(connectContext,
                     () -> executeQueryStatement(context.peerIdentity(), connectContext, request.getQuery(),
                             descriptor));
+        } catch (FlightRuntimeException e) {
+            // Already carries the status meant for the client - UNAVAILABLE from the session's
+            // command lock or from an incremental window that is not ready (queryFailure, with its
+            // doris-error-code metadata), UNAUTHENTICATED from a closed session, and whatever the
+            // session layer refuses a session with. Wrapping it as INTERNAL would hide that, as it
+            // did between #67820 and this fix; the other entry points below let it through too.
+            throw e;
         } catch (Throwable e) {
-            if (e instanceof FlightRuntimeException) {
-                FlightRuntimeException flightError = (FlightRuntimeException) e;
-                ErrorFlightMetadata metadata = flightError.status().metadata();
-                if (metadata.containsKey("doris-error-code")) {
-                    String code = metadata.get("doris-error-code");
-                    if (Integer.toString(ErrorCode.ERR_INCR_WINDOW_NOT_READY.getCode()).equals(code)
-                            || Integer.toString(ErrorCode.ERR_INCR_VISIBLE_WAIT_TIMEOUT.getCode()).equals(code)) {
-                        throw flightError;
-                    }
-                }
-            }
             String errMsg = "get flight info statement failed, " + e.getMessage();
             LOG.error(errMsg, e);
             throw CallStatus.INTERNAL.withDescription(errMsg).withCause(e).toRuntimeException();

fe/fe-core/src/test/java/org/apache/doris/arrowflight/DorisFlightSqlProducerTest.java

@@ -107,29 +107,37 @@ public void testGetFlightInfoPreservesBothWindowErrors() throws Exception {
         }
     }
 
+    // A Flight status chosen by the session layer reaches the client as is, whatever it is: the
+    // session's command lock (UNAVAILABLE), a closed session (UNAUTHENTICATED), a refused session
+    // (RESOURCE_EXHAUSTED). Only a non-Flight failure is wrapped as INTERNAL.
     @Test
-    public void testGetFlightInfoWrapsOtherFlightErrors() throws Exception {
+    public void testGetFlightInfoPassesOtherFlightErrorsThrough() throws Exception {
         for (CallStatus status : new CallStatus[] {CallStatus.INTERNAL, CallStatus.UNAVAILABLE,
-                CallStatus.INVALID_ARGUMENT, CallStatus.UNAUTHENTICATED}) {
+                CallStatus.INVALID_ARGUMENT, CallStatus.UNAUTHENTICATED, CallStatus.RESOURCE_EXHAUSTED}) {
             FlightRuntimeException failure = status.withDescription("other flight failure").toRuntimeException();
-            assertLegacyFlightWrapper(failure, getFlightInfoFailure(failure));
+            Assertions.assertSame(failure, getFlightInfoFailure(failure));
         }
     }
 
     @Test
-    public void testGetFlightInfoWrapsOtherBusinessErrors() throws Exception {
+    public void testGetFlightInfoPassesOtherBusinessErrorsThrough() throws Exception {
         ErrorFlightMetadata metadata = new ErrorFlightMetadata();
         metadata.insert("doris-error-code", Integer.toString(ErrorCode.ERR_UNKNOWN_ERROR.getCode()));
         FlightRuntimeException failure = CallStatus.UNAVAILABLE.withDescription("other business failure")
                 .withMetadata(metadata).toRuntimeException();
 
-        assertLegacyFlightWrapper(failure, getFlightInfoFailure(failure));
+        Assertions.assertSame(failure, getFlightInfoFailure(failure));
     }
 
     @Test
     public void testGetFlightInfoWrapsNonFlightErrors() throws Exception {
         RuntimeException failure = new RuntimeException("session lookup failed");
-        assertLegacyFlightWrapper(failure, getFlightInfoFailure(failure));
+        FlightRuntimeException result = getFlightInfoFailure(failure);
+        Assertions.assertEquals(FlightStatusCode.INTERNAL, result.status().code());
+        Assertions.assertSame(failure, result.getCause());
+        Assertions.assertEquals("get flight info statement failed, " + failure.getMessage(),
+                result.status().description());
+        Assertions.assertFalse(result.status().metadata().containsKey("doris-error-code"));
     }
 
     private FlightRuntimeException getFlightInfoFailure(RuntimeException failure) throws Exception {
@@ -145,15 +153,6 @@ private FlightRuntimeException getFlightInfoFailure(RuntimeException failure) th
         }
     }
 
-    private void assertLegacyFlightWrapper(RuntimeException failure, FlightRuntimeException result) {
-        Assertions.assertEquals(FlightStatusCode.INTERNAL, result.status().code());
-        Assertions.assertNotSame(failure, result);
-        Assertions.assertSame(failure, result.getCause());
-        Assertions.assertEquals("get flight info statement failed, " + failure.getMessage(),
-                result.status().description());
-        Assertions.assertFalse(result.status().metadata().containsKey("doris-error-code"));
-    }
-
     @BeforeEach
     public void setUp() {
         // ConnectContext.init() only reaches Env when this is false; keep it true so the