HIVE-29818: ServletSecurity needs a UGI cache

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -1980,6 +1980,17 @@ public enum ConfVars {
         "hive.metastore.iceberg.catalog.metrics.reporters", "org.apache.iceberg.rest.metrics.LoggingMetricsReporter",
         "A comma separated list of custom Iceberg Metrics Reporting plugins."
     ),
+    CATALOG_SERVLET_UGI_CACHE_SIZE("metastore.catalog.servlet.ugi.cache.size",
+        "hive.metastore.catalog.servlet.ugi.cache.size", 1000L,
+        "Maximum number of proxy UserGroupInformation instances to keep in the catalog servlet UGI cache. " +
+        "Entries displaced by this limit trigger FileSystem resource cleanup for the evicted UGI."
+    ),
+    CATALOG_SERVLET_UGI_CACHE_EXPIRY("metastore.catalog.servlet.ugi.cache.expiry",
+        "hive.metastore.catalog.servlet.ugi.cache.expiry", 3600, TimeUnit.SECONDS,
+        "Idle-expiry time for cached proxy UserGroupInformation instances in the catalog servlet. " +
+        "After this period of inactivity, the entry is evicted and FileSystem.closeAllForUGI is called " +
+        "to release associated IPC and RPC resources. Set to 0 to disable expiry-based eviction."
+    ),
     HTTPSERVER_THREADPOOL_MIN("hive.metastore.httpserver.threadpool.min",
             "hive.metastore.httpserver.threadpool.min", 8,
             "HMS embedded HTTP server minimum number of threads."

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java

@@ -21,12 +21,22 @@
 
 import static javax.ws.rs.core.HttpHeaders.WWW_AUTHENTICATE;
 
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.Ticker;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import java.security.KeyStore;
+import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.hive.metastore.auth.HttpAuthenticationException;
 import org.apache.hadoop.hive.metastore.auth.jwt.SimpleJWTAuthenticator;
 import org.apache.hadoop.hive.metastore.auth.oauth2.OAuth2Authenticator;
@@ -106,17 +116,113 @@
   private final Function<HttpServletRequest, List<String>> scopeProvider;
   private SimpleJWTAuthenticator jwtAuthenticator = null;
   private OAuth2Authenticator oAuth2Authenticator = null;
+  private final Cache<UgiKey, UserGroupInformation> proxyUserCache;
+
+  /**
+   * Cache key for a proxy {@link UserGroupInformation}. A proxy UGI is bound to both the effective user it
+   * impersonates and the server login user acting as its real user, so both participate in identity.
+   */
+  record UgiKey(String effectiveUser, String loginUser) {}
 
   public ServletSecurity(AuthType authType, Configuration conf) {
     this(authType, conf, null);
   }
 
   public ServletSecurity(AuthType authType, Configuration conf,
       Function<HttpServletRequest, List<String>> scopeProvider) {
+    this(authType, conf, scopeProvider, ForkJoinPool.commonPool());
+  }
+
+  @VisibleForTesting
+  ServletSecurity(AuthType authType, Configuration conf,
+      Function<HttpServletRequest, List<String>> scopeProvider, Executor cacheCleanupExecutor) {
     this.conf = conf;
     this.isSecurityEnabled = UserGroupInformation.isSecurityEnabled();
     this.authType = authType;
     this.scopeProvider = scopeProvider;
+    this.proxyUserCache = createCacheWithConfig(
+        MetastoreConf.getTimeVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_EXPIRY, TimeUnit.MILLISECONDS),
+        MetastoreConf.getLongVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE),
+        cacheCleanupExecutor);
+  }
+
+  /**
+   * Creates a UGI cache with the specified expiration time and maximum size.
+   *
+   * @param expirationMs Time in milliseconds after which entries expire due to inactivity
+   * @param maxSize      Maximum number of entries the cache can hold
+   * @param cacheCleanupExecutor executor on which removal-listener cleanup ({@link FileSystem#closeAllForUGI})
+   *                             runs; production uses {@link ForkJoinPool#commonPool()} so cleanup stays off the
+   *                             request thread
+   * @return A configured Caffeine cache for UGI objects
+   */
+  private Cache<UgiKey, UserGroupInformation> createCacheWithConfig(long expirationMs, long maxSize,
+      Executor cacheCleanupExecutor) {
+    // Note: eviction closes the UGI's FileSystems. If an entry is evicted while a request is still inside doAs,
+    // that in-flight operation could see a "FileSystem closed" error. We don't reference-count to prevent this;
+    // instead we rely on generous margins: expiry is idle-based (expireAfterAccess), and both the expiry window
+    // and maximumSize should be kept well above the longest operation / peak concurrent distinct users.
+    RemovalListener<UgiKey, UserGroupInformation> cleanupListener =
+        (key, ugi, cause) -> {
+          if (ugi != null) {
+            try {
+              FileSystem.closeAllForUGI(ugi);
+              if (LOG.isDebugEnabled()) {
+                LOG.debug("Cleaned up FileSystem handles for evicted UGI: {} (cause: {})",
+                    ugi.getUserName(), cause);
+              }
+            } catch (IOException cleanupException) {
+              LOG.error("Failed to clean up FileSystem handles for evicted UGI: {} (cause: {})",
+                  ugi, cause, cleanupException);
+            }
+          }
+        };
+
+    Caffeine<UgiKey, UserGroupInformation> builder = Caffeine.<UgiKey, UserGroupInformation>newBuilder()
+        .maximumSize(maxSize)
+        .executor(cacheCleanupExecutor)
+        .removalListener(cleanupListener);
+
+    if (expirationMs > 0) {
+      builder.expireAfterAccess(Duration.ofMillis(expirationMs))
+          .ticker(Ticker.systemTicker());
+    }
+
+    return builder.build();
+  }
+
+  /**
+   * Returns the (cached) proxy {@link UserGroupInformation} for the given user, creating it on a cache miss.
+   * <p>Caching the proxy UGI prevents Hadoop{@literal '}s {@code FileSystem.CACHE} from accumulating a distinct
+   * entry (and its RPC/IPC resources) for every request; evicted entries are cleaned up through
+   * {@link FileSystem#closeAllForUGI(UserGroupInformation)}.</p>
+   * @param userName the effective user name extracted from the request
+   * @param loginUser the server login user that acts as the real user of the proxy
+   * @return the cached or freshly created proxy user
+   */
+  @VisibleForTesting
+  UserGroupInformation getProxyUser(String userName, UserGroupInformation loginUser) {
+    return proxyUserCache.get(new UgiKey(userName, loginUser.getUserName()), key -> {
+      LOG.debug("Creating proxy user for: {}", key.effectiveUser());
+      return UserGroupInformation.createProxyUser(key.effectiveUser(), loginUser);
+    });
+  }
+
+  /**
+   * Forces the proxy user cache to run any pending maintenance (eviction and cleanup).
+   */
+  @VisibleForTesting
+  void cleanUpProxyUserCache() {
+    proxyUserCache.cleanUp();
+  }
+
+  /**
+   * @return the approximate number of entries currently held in the proxy user cache
+   */
+  @VisibleForTesting
+  long proxyUserCacheSize() {
+    proxyUserCache.cleanUp();
+    return proxyUserCache.estimatedSize();
   }
 
   /**
@@ -237,8 +343,7 @@
       // Temporary, and useless for now. Here only to allow this to work on an otherwise kerberized
       // server.
       if (isSecurityEnabled || authType == AuthType.JWT || authType == AuthType.OAUTH2) {
-        LOG.info("Creating proxy user for: {}", userFromHeader);
-        clientUgi = UserGroupInformation.createProxyUser(userFromHeader, UserGroupInformation.getLoginUser());
+        clientUgi = getProxyUser(userFromHeader, UserGroupInformation.getLoginUser());
       } else {
         // Unreachable in the case of NONE
         Preconditions.checkState(authType == AuthType.SIMPLE);

standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java

@@ -0,0 +1,102 @@
+/*
+ * 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.hadoop.hive.metastore;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+/**
+ * Unit tests for the proxy {@link UserGroupInformation} cache used by {@link ServletSecurity} to bound the number of
+ * proxy UGIs (and their associated FileSystem resources) that would otherwise leak through Hadoop's FileSystem cache.
+ */
+@Category(MetastoreUnitTest.class)
+public class TestServletSecurity {
+
+  private static Configuration confWithCache(long maxSize, long expirySeconds) {
+    Configuration conf = MetastoreConf.newMetastoreConf();
+    MetastoreConf.setLongVar(conf, ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE, maxSize);
+    MetastoreConf.setTimeVar(conf, ConfVars.CATALOG_SERVLET_UGI_CACHE_EXPIRY, expirySeconds, TimeUnit.SECONDS);
+    return conf;
+  }
+
+  @Test
+  public void testProxyUserIsCachedPerUser() throws Exception {
+    ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 3600));
+    UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
+
+    UserGroupInformation first = security.getProxyUser("alice", loginUser);
+    UserGroupInformation second = security.getProxyUser("alice", loginUser);
+
+    Assert.assertSame("Repeated requests for the same user must reuse the cached proxy UGI", first, second);
+    Assert.assertEquals("alice", first.getShortUserName());
+  }
+
+  @Test
+  public void testDistinctUsersGetDistinctProxies() throws Exception {
+    ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 3600));
+    UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
+
+    UserGroupInformation alice = security.getProxyUser("alice", loginUser);
+    UserGroupInformation bob = security.getProxyUser("bob", loginUser);
+
+    Assert.assertNotSame(alice, bob);
+    Assert.assertEquals(2, security.proxyUserCacheSize());
+  }
+
+  @Test
+  public void testEvictionClosesFileSystemForUgi() throws Exception {
+    // A cache bounded to a single entry: inserting a second user evicts the first.
+    // Runnable::run makes removal-listener cleanup run synchronously on this thread, where the static mock is active.
+    ServletSecurity security =
+        new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(1, 3600), null, Runnable::run);
+    UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
+
+    try (MockedStatic<FileSystem> fsMock = Mockito.mockStatic(FileSystem.class)) {
+      UserGroupInformation alice = security.getProxyUser("alice", loginUser);
+      security.getProxyUser("bob", loginUser);
+      // Force any pending size-based eviction (and its synchronous cleanup) to run.
+      security.cleanUpProxyUserCache();
+
+      fsMock.verify(() -> FileSystem.closeAllForUGI(alice));
+    }
+  }
+
+  @Test
+  public void testExpiryDisabledWhenNonPositive() throws Exception {
+    // expiry == 0 disables time-based eviction; the size bound still applies and entries remain until displaced.
+    ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(100, 0));
+    UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
+
+    UserGroupInformation first = security.getProxyUser("carol", loginUser);
+    UserGroupInformation second = security.getProxyUser("carol", loginUser);
+
+    Assert.assertSame(first, second);
+  }
+}