Copilot commented on code in PR #8189: URL: https://github.com/apache/gravitino/pull/8189#discussion_r2579308948
########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapperProxy.java: ########## @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.iceberg.common.ops; + +import java.lang.reflect.Method; +import net.sf.cglib.proxy.Enhancer; +import net.sf.cglib.proxy.MethodInterceptor; +import net.sf.cglib.proxy.MethodProxy; +import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; +import org.apache.iceberg.catalog.Catalog; + Review Comment: Missing class-level Javadoc. This proxy class plays a critical role in handling Kerberos authentication and impersonation. Add comprehensive documentation explaining its purpose, how it intercepts method calls, and when/why it should be used. ```suggestion /** * Proxy class for {@link IcebergCatalogWrapper} that intercepts method calls to handle Kerberos authentication * and impersonation. This class uses CGLIB's {@link MethodInterceptor} to wrap method invocations on the * target catalog, ensuring that any operations requiring Kerberos credentials are executed within the correct * security context. * * <p>When a method is invoked on the proxy, if the underlying {@link Catalog} supports Kerberos (i.e., implements * {@link SupportsKerberos}), the call is wrapped in a Kerberos operation using the provided configuration * properties. Otherwise, the method is invoked directly on the target. * * <p>This proxy should be used whenever an {@link IcebergCatalogWrapper} may interact with a Kerberos-secured * Iceberg catalog, to ensure proper authentication and impersonation are handled transparently. */ ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -93,4 +266,18 @@ private void closeInternalClientPool() { LOGGER.warn("Failed to close HiveCatalog internal client pool", e); } } + + private KerberosClient initKerberosClient() { + try { + KerberosClient kerberosClient = new KerberosClient(this.properties(), this.getConf()); + // For Iceberg rest server, we haven't set the catalog_uuid, so we set it to 0 as there is + // only one catalog in the rest server, so it's okay to set it to 0. + String catalogUUID = properties().getOrDefault("catalog_uuid", "0"); + File keytabFile = kerberosClient.saveKeyTabFileFromUri(Long.valueOf(catalogUUID)); Review Comment: Use `Long.parseLong()` instead of `Long.valueOf()` with wrapper boxing. The `saveKeyTabFileFromUri()` expects a `long` primitive, so `Long.parseLong()` is more direct and avoids unnecessary object creation. Also, this can throw `NumberFormatException` if the `catalog_uuid` is not a valid long, which should be handled or documented. ```suggestion /** * Initializes the Kerberos client and logs in using the keytab file. * * @return the initialized KerberosClient * @throws NumberFormatException if the catalog_uuid property is not a valid long */ private KerberosClient initKerberosClient() { try { KerberosClient kerberosClient = new KerberosClient(this.properties(), this.getConf()); // For Iceberg rest server, we haven't set the catalog_uuid, so we set it to 0 as there is // only one catalog in the rest server, so it's okay to set it to 0. String catalogUUID = properties().getOrDefault("catalog_uuid", "0"); File keytabFile = kerberosClient.saveKeyTabFileFromUri(Long.parseLong(catalogUUID)); ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -68,6 +112,135 @@ public void close() throws IOException { }); } + @Override + public <R> R doKerberosOperations(Map<String, String> properties, Executable<R> executable) + throws Throwable { + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (!authenticationConfig.isKerberosAuth()) { + return executable.execute(); + } + + final String finalPrincipalName; + String proxyKerberosPrincipalName = PrincipalUtils.getCurrentPrincipal().getName(); + + if (!proxyKerberosPrincipalName.contains("@")) { + finalPrincipalName = + String.format("%s@%s", proxyKerberosPrincipalName, kerberosClient.getRealm()); + } else { + finalPrincipalName = proxyKerberosPrincipalName; + } + + UserGroupInformation realUser = + authenticationConfig.isImpersonationEnabled() + ? UserGroupInformation.createProxyUser( + finalPrincipalName, kerberosClient.getLoginUser()) + : kerberosClient.getLoginUser(); + try { + ClientPool<IMetaStoreClient, TException> newClientPool = + (ClientPool<IMetaStoreClient, TException>) FieldUtils.readField(this, "clients", true); + kerberosClient + .getLoginUser() + .doAs( + (PrivilegedExceptionAction<Void>) + () -> { + String token = + newClientPool.run( + client -> + client.getDelegationToken( + finalPrincipalName, + kerberosClient.getLoginUser().getShortUserName())); + + Token<DelegationTokenIdentifier> delegationToken = new Token<>(); + delegationToken.decodeFromUrlString(token); + realUser.addToken(delegationToken); + return null; + }); + } catch (Exception e) { + throw new RuntimeException("Failed to get delegation token", e); Review Comment: The error message "Failed to get delegation token" doesn't include the exception details or context about which principal was being used. Consider including `finalPrincipalName` and the original exception message in the RuntimeException to aid debugging: `new RuntimeException("Failed to get delegation token for principal: " + finalPrincipalName, e)`. ```suggestion throw new RuntimeException("Failed to get delegation token for principal: " + finalPrincipalName, e); ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -68,6 +112,135 @@ public void close() throws IOException { }); } + @Override + public <R> R doKerberosOperations(Map<String, String> properties, Executable<R> executable) + throws Throwable { + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (!authenticationConfig.isKerberosAuth()) { + return executable.execute(); + } + + final String finalPrincipalName; + String proxyKerberosPrincipalName = PrincipalUtils.getCurrentPrincipal().getName(); + + if (!proxyKerberosPrincipalName.contains("@")) { + finalPrincipalName = + String.format("%s@%s", proxyKerberosPrincipalName, kerberosClient.getRealm()); + } else { + finalPrincipalName = proxyKerberosPrincipalName; + } + + UserGroupInformation realUser = + authenticationConfig.isImpersonationEnabled() + ? UserGroupInformation.createProxyUser( + finalPrincipalName, kerberosClient.getLoginUser()) + : kerberosClient.getLoginUser(); Review Comment: Potential NullPointerException if `kerberosClient` is null when Kerberos authentication is enabled. The method checks `authenticationConfig.isKerberosAuth()` at line 120, but if `initKerberosClient()` failed during initialization (line 76), `kerberosClient` will be null. This would cause NPE when accessing `kerberosClient.getRealm()` at line 129 or `kerberosClient.getLoginUser()` at line 137. Add a null check for `kerberosClient` or ensure initialization always succeeds or throws an exception. ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/authentication/SupportsKerberos.java: ########## @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.iceberg.common.authentication; + +import java.util.Map; + +/** + * An interface to indicate that the implementing class supports Kerberos authentication and can do + * operations with Kerberos. + */ +public interface SupportsKerberos { + + /** + * Perform operations with Kerberos authentication. + * + * @param properties the catalog properties + * @param executable the operations to be performed + * @return the result of the operations + * @param <R> the return type of the operations Review Comment: The method documentation is missing the `@throws` annotation. The method signature declares `throws Throwable`, which should be documented to explain what exceptions callers should expect to handle. ```suggestion * @param <R> the return type of the operations * @throws Throwable if an error occurs during Kerberos authentication or if the provided * {@link Executable} throws an exception ``` ########## catalogs/catalog-lakehouse-iceberg/src/test/java/org/apache/gravitino/catalog/lakehouse/iceberg/integration/test/CatalogIcebergKerberosHiveIT.java: ########## @@ -262,6 +262,33 @@ void testIcebergWithKerberosAndUserImpersonation() throws IOException { Assertions.assertDoesNotThrow( () -> catalog.asSchemas().createSchema(SCHEMA_NAME, "comment", ImmutableMap.of())); + kerberosHiveContainer.executeInContainer( + "hadoop", "fs", "-chmod", "-R", "000", "/user/hive/warehouse-catalog-iceberg"); + + // Create table and the user has no permission to create table Review Comment: The comment "Create table and the user has no permission to create table" is slightly misleading because the test is actually verifying that table creation fails when the user lacks EXECUTE permission on the parent directory (as shown by the assertion at line 287). Consider updating the comment to be more specific: "Create table when the user has no EXECUTE permission on the warehouse directory". ```suggestion // Create table when the user has no EXECUTE permission on the warehouse directory ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapperProxy.java: ########## @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.iceberg.common.ops; + +import java.lang.reflect.Method; +import net.sf.cglib.proxy.Enhancer; +import net.sf.cglib.proxy.MethodInterceptor; +import net.sf.cglib.proxy.MethodProxy; +import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; +import org.apache.iceberg.catalog.Catalog; + +public class IcebergCatalogWrapperProxy implements MethodInterceptor { + private final IcebergCatalogWrapper target; + private final Catalog catalog; + + public IcebergCatalogWrapperProxy(IcebergCatalogWrapper target) { + this.target = target; + this.catalog = target.catalog; + } + + @Override + public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) + throws Throwable { + if (catalog instanceof SupportsKerberos) { + SupportsKerberos kerberosCatalog = (SupportsKerberos) catalog; + return kerberosCatalog.doKerberosOperations( + target.getIcebergConfig().getIcebergCatalogProperties(), + () -> methodProxy.invoke(target, objects)); + } + + return method.invoke(target, objects); + } + + public IcebergCatalogWrapper getProxy(IcebergConfig config) { + Enhancer e = new Enhancer(); + e.setClassLoader(target.getClass().getClassLoader()); + e.setSuperclass(target.getClass()); + e.setCallback(this); + + Class<?>[] argClass = new Class[] {IcebergConfig.class}; + return (IcebergCatalogWrapper) e.create(argClass, new Object[] {config}); + } Review Comment: The proxy creation might fail if `IcebergCatalogWrapper` doesn't have a public constructor matching `IcebergConfig` parameter. While the code appears to work, the method doesn't handle potential exceptions from `e.create()`. Consider wrapping in a try-catch block to provide a more informative error message if proxy creation fails, or add documentation about the expected constructor signature. ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -68,6 +112,135 @@ public void close() throws IOException { }); } + @Override + public <R> R doKerberosOperations(Map<String, String> properties, Executable<R> executable) + throws Throwable { + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (!authenticationConfig.isKerberosAuth()) { + return executable.execute(); + } + + final String finalPrincipalName; + String proxyKerberosPrincipalName = PrincipalUtils.getCurrentPrincipal().getName(); + + if (!proxyKerberosPrincipalName.contains("@")) { + finalPrincipalName = + String.format("%s@%s", proxyKerberosPrincipalName, kerberosClient.getRealm()); + } else { + finalPrincipalName = proxyKerberosPrincipalName; + } + + UserGroupInformation realUser = + authenticationConfig.isImpersonationEnabled() + ? UserGroupInformation.createProxyUser( + finalPrincipalName, kerberosClient.getLoginUser()) + : kerberosClient.getLoginUser(); + try { + ClientPool<IMetaStoreClient, TException> newClientPool = + (ClientPool<IMetaStoreClient, TException>) FieldUtils.readField(this, "clients", true); Review Comment: Unsafe unchecked cast from `Object` to `ClientPool<IMetaStoreClient, TException>`. Consider adding a type check or catching `ClassCastException` to provide a more informative error message if the field type is unexpected. ```suggestion Object clientsField = FieldUtils.readField(this, "clients", true); if (!(clientsField instanceof ClientPool)) { throw new RuntimeException( "Expected field 'clients' to be of type ClientPool, but got: " + (clientsField == null ? "null" : clientsField.getClass().getName())); } @SuppressWarnings("unchecked") ClientPool<IMetaStoreClient, TException> newClientPool = (ClientPool<IMetaStoreClient, TException>) clientsField; ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -47,8 +67,32 @@ public void addResource(Closeable resource) { resources.add(resource); } + @Override + public void initialize(String inputName, Map<String, String> properties) { + super.initialize(inputName, properties); + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (authenticationConfig.isKerberosAuth()) { + this.kerberosClient = initKerberosClient(); + } + + try { + resetIcebergHiveClientPool(); Review Comment: The `resetIcebergHiveClientPool()` method is called during initialization and replaces the client pool using reflection. If `initialize()` can be called multiple times or from multiple threads, this could lead to race conditions or resource leaks (the old pool might be closed while still in use). Consider adding synchronization or documenting that `initialize()` must be called only once. ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -68,6 +112,135 @@ public void close() throws IOException { }); } + @Override + public <R> R doKerberosOperations(Map<String, String> properties, Executable<R> executable) + throws Throwable { + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (!authenticationConfig.isKerberosAuth()) { + return executable.execute(); + } + + final String finalPrincipalName; + String proxyKerberosPrincipalName = PrincipalUtils.getCurrentPrincipal().getName(); + + if (!proxyKerberosPrincipalName.contains("@")) { + finalPrincipalName = + String.format("%s@%s", proxyKerberosPrincipalName, kerberosClient.getRealm()); + } else { + finalPrincipalName = proxyKerberosPrincipalName; + } + + UserGroupInformation realUser = + authenticationConfig.isImpersonationEnabled() + ? UserGroupInformation.createProxyUser( + finalPrincipalName, kerberosClient.getLoginUser()) + : kerberosClient.getLoginUser(); + try { + ClientPool<IMetaStoreClient, TException> newClientPool = + (ClientPool<IMetaStoreClient, TException>) FieldUtils.readField(this, "clients", true); + kerberosClient + .getLoginUser() + .doAs( + (PrivilegedExceptionAction<Void>) + () -> { + String token = + newClientPool.run( + client -> + client.getDelegationToken( + finalPrincipalName, + kerberosClient.getLoginUser().getShortUserName())); + + Token<DelegationTokenIdentifier> delegationToken = new Token<>(); + delegationToken.decodeFromUrlString(token); + realUser.addToken(delegationToken); + return null; + }); + } catch (Exception e) { + throw new RuntimeException("Failed to get delegation token", e); + } + return (R) Review Comment: Unsafe unchecked cast from `Object` to `R`. The `doAs` method returns `Object`, and casting it directly to generic type `R` without any runtime type checking can lead to `ClassCastException`. Consider using a type token pattern or document that the caller is responsible for type safety. ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapperProxy.java: ########## @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.gravitino.iceberg.common.ops; + +import java.lang.reflect.Method; +import net.sf.cglib.proxy.Enhancer; +import net.sf.cglib.proxy.MethodInterceptor; +import net.sf.cglib.proxy.MethodProxy; +import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; +import org.apache.iceberg.catalog.Catalog; + +public class IcebergCatalogWrapperProxy implements MethodInterceptor { + private final IcebergCatalogWrapper target; + private final Catalog catalog; + + public IcebergCatalogWrapperProxy(IcebergCatalogWrapper target) { + this.target = target; + this.catalog = target.catalog; Review Comment: Direct field access `target.catalog` bypasses encapsulation. The `catalog` field is protected in `IcebergCatalogWrapper`, and while this works because they're in the same package, it's better to use the getter method `target.getCatalog()` for better encapsulation and consistency. ```suggestion this.catalog = target.getCatalog(); ``` ########## gradle/libs.versions.toml: ########## @@ -92,7 +92,8 @@ servlet = "3.1.0" jodd = "3.5.2" flink = "1.18.0" flinkjdbc = "3.2.0-1.18" Review Comment: The cglib version is upgraded from 2.2 to 3.3.0, which is a significant version jump. While the ASM exclusion and explicit dependency addition are necessary for Java 17 compatibility, consider documenting this version change in the PR description or commit message to help with future debugging if any compatibility issues arise. The change appears correct, but the justification could be more visible. ```suggestion flinkjdbc = "3.2.0-1.18" # cglib upgraded from 2.2 to 3.3.0 for Java 17 compatibility. # Explicit ASM dependency (9.8) is required due to cglib's internal use and Java 17 module system. ``` ########## catalogs/catalog-lakehouse-iceberg/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergCatalogOperations.java: ########## @@ -113,7 +115,18 @@ public void initialize( resultConf.put("catalog_uuid", info.id().toString()); IcebergConfig icebergConfig = new IcebergConfig(resultConf); - this.icebergCatalogWrapper = new IcebergCatalogWrapper(icebergConfig); + IcebergCatalogWrapper rawWrapper = new IcebergCatalogWrapper(icebergConfig); + // We have replaced `UserGroupInformation#loginUserFromKeytab` with + // `UserGroupInformation#loginUserFromKeytabAndReturnUGI`, the former will change the current + // login user globally, which is not expected in Gravitino as we are going to support multiple + // catalogs within the same class loader. The proxy will ensure each catalog has its own + // `UserGroupInformation` instance and I have removed old `HiveBackendProxy`. In + // IcebergCatalogWrapperProxy, we will do both Kerberos access and user impersonation if needed, + // so please check the code in IcebergCatalogWrapperProxy for details. Review Comment: The comment states "We have replaced `UserGroupInformation#loginUserFromKeytab` with `UserGroupInformation#loginUserFromKeytabAndReturnUGI`", but the actual code in `KerberosClient.java` line 77 still uses `UserGroupInformation.loginUserFromKeytab()`. This inconsistency between documentation and implementation is misleading and should be corrected. Either update the code to use `loginUserFromKeytabAndReturnUGI` if that was the intent, or update the comment to reflect the actual implementation. ```suggestion // Kerberos authentication and user impersonation are handled within IcebergCatalogWrapperProxy. // The proxy ensures that each catalog has its own UserGroupInformation instance, avoiding // global changes to the login user, which is important for supporting multiple catalogs within // the same class loader. For details on how Kerberos access and user impersonation are managed, // please refer to the code in IcebergCatalogWrapperProxy. ``` ########## iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ClosableHiveCatalog.java: ########## @@ -47,8 +67,32 @@ public void addResource(Closeable resource) { resources.add(resource); } + @Override + public void initialize(String inputName, Map<String, String> properties) { + super.initialize(inputName, properties); + + AuthenticationConfig authenticationConfig = new AuthenticationConfig(properties); + if (authenticationConfig.isKerberosAuth()) { + this.kerberosClient = initKerberosClient(); + } + + try { + resetIcebergHiveClientPool(); + } catch (Exception e) { + throw new RuntimeException("Failed to reset IcebergHiveClientPool", e); + } Review Comment: If `initKerberosClient()` throws an exception at line 76, the `kerberosClient` remains null, but `resetIcebergHiveClientPool()` is still called at line 80. This could leave the catalog in an inconsistent state where Kerberos authentication is expected but not properly initialized. Consider failing fast if Kerberos initialization fails when it's required, or ensuring `resetIcebergHiveClientPool()` can handle a null `kerberosClient`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
