[PAIMON-9337] Support Kerberos proxy user when reading Paimon table via Hive

paimon-common/src/main/java/org/apache/paimon/security/KerberosLoginProvider.java

@@ -73,7 +73,8 @@ public boolean isLoginPossible() throws IOException {
                 return true;
             }
         } else {
-            throwProxyUserNotSupported();
+            LOG.debug("Login from ProxyUser");
+            return true;
         }
 
         LOG.debug("Login is NOT possible");
@@ -95,14 +96,11 @@ public void doLogin() throws IOException {
             UserGroupInformation.loginUserFromSubject(null);
             LOG.info("Loaded user's ticket cache successfully");
         } else {
-            throwProxyUserNotSupported();
+            // Proxy user: credentials are already provided by the proxy mechanism.
+            LOG.debug("Skipping Kerberos login for proxy user");
         }
     }
 
-    private void throwProxyUserNotSupported() {
-        throw new UnsupportedOperationException("Proxy user is not supported");
-    }
-
     public static boolean isProxyUser(UserGroupInformation ugi) {
         return ugi.getAuthenticationMethod() == UserGroupInformation.AuthenticationMethod.PROXY;
     }

paimon-common/src/test/java/org/apache/paimon/security/KerberosLoginProviderITCase.java

@@ -33,10 +33,10 @@
 import static org.apache.paimon.security.SecurityConfiguration.KERBEROS_LOGIN_PRINCIPAL;
 import static org.apache.paimon.security.SecurityConfiguration.KERBEROS_LOGIN_USETICKETCACHE;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.when;
 
 /**
@@ -108,7 +108,7 @@
     }
 
     @Test
-    public void isLoginPossibleMustThrowExceptionWithProxyUser() {
+    public void isLoginPossibleMustReturnTrueWithProxyUser() {
         Options options = new Options();
         KerberosLoginProvider kerberosLoginProvider = new KerberosLoginProvider(options);
 
@@ -119,8 +119,7 @@
             ugi.when(UserGroupInformation::isSecurityEnabled).thenReturn(true);
             ugi.when(UserGroupInformation::getCurrentUser).thenReturn(userGroupInformation);
 
-            assertThatThrownBy(kerberosLoginProvider::isLoginPossible)
-                    .isInstanceOf(UnsupportedOperationException.class);
+            assertThat(kerberosLoginProvider.isLoginPossible()).isTrue();
         }
     }
 
@@ -158,7 +157,7 @@
     }
 
     @Test
-    public void doLoginMustThrowExceptionWithProxyUser() {
+    public void doLoginMustDoNothingWithProxyUser() {
         Options options = new Options();
         KerberosLoginProvider kerberosLoginProvider = new KerberosLoginProvider(options);
 
@@ -168,8 +167,12 @@
                     .thenReturn(UserGroupInformation.AuthenticationMethod.PROXY);
             ugi.when(UserGroupInformation::getCurrentUser).thenReturn(userGroupInformation);
 
-            assertThatThrownBy(kerberosLoginProvider::doLogin)
-                    .isInstanceOf(UnsupportedOperationException.class);
+            // proxy user should not trigger any Kerberos login
+            kerberosLoginProvider.doLogin();
+            ugi.verify(() -> UserGroupInformation.loginUserFromSubject(null), never());
+            ugi.verify(
+                    () -> UserGroupInformation.loginUserFromKeytab(anyString(), anyString()),
+                    never());
         }
     }
 }