From 7859fe335adcb83459feb7254320d5fffcb0b5c7 Mon Sep 17 00:00:00 2001 From: Henrib Date: Mon, 17 Aug 2026 17:34:40 +0200 Subject: [PATCH 1/2] HIVE-29818: ServletSecurity needs a UGI cache The REST Catalog creates a fresh proxy UserGroupInformation per request via UserGroupInformation.createProxyUser. Hadoop's FileSystem.CACHE retains a reference to every such UGI (and its RPC/IPC resources), so under proxy authentication these short-lived UGIs accumulate and eventually exhaust memory in long-running deployments. Cache the proxy UGI in ServletSecurity with a bounded, idle-evicting Caffeine cache keyed by (realUser, loginUser). Evicted entries release their resources via FileSystem.closeAllForUGI. Two new config vars tune the cache: - metastore.catalog.servlet.ugi.cache.size (default 1000) - metastore.catalog.servlet.ugi.cache.expiry (default 3600s, 0 disables) Add TestServletSecurity covering per-user caching, distinct proxies per user, eviction-triggered FileSystem cleanup, and disabled expiry. --- .../hive/metastore/conf/MetastoreConf.java | 11 ++ .../hive/metastore/ServletSecurity.java | 96 ++++++++++++++++- .../hive/metastore/TestServletSecurity.java | 100 ++++++++++++++++++ 3 files changed, 205 insertions(+), 2 deletions(-) create mode 100644 standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java index accc6c4b4dff..96f1db1db2b6 100644 --- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java +++ b/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." diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java index 944c490e9787..13a3d2acb9af 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java @@ -21,12 +21,20 @@ 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.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,6 +114,13 @@ public static AuthType fromString(String type) { private final Function> scopeProvider; private SimpleJWTAuthenticator jwtAuthenticator = null; private OAuth2Authenticator oAuth2Authenticator = null; + private final Cache 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 realUser, String loginUser) {} public ServletSecurity(AuthType authType, Configuration conf) { this(authType, conf, null); @@ -117,6 +132,84 @@ public ServletSecurity(AuthType authType, Configuration 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), + (int) MetastoreConf.getLongVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE)); + } + + /** + * 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 + * @return A configured Caffeine cache for UGI objects + */ + private Cache createCacheWithConfig(long expirationMs, int maxSize) { + // 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 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 builder = Caffeine.newBuilder() + .maximumSize(maxSize) + .executor(Runnable::run) + .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. + *

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)}.

+ * @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.info("Creating proxy user for: {}", key.realUser()); + return UserGroupInformation.createProxyUser(key.realUser(), 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 +330,7 @@ public void execute(HttpServletRequest request, HttpServletResponse response, Me // 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); diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java new file mode 100644 index 000000000000..e91b1907ffbe --- /dev/null +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java @@ -0,0 +1,100 @@ +/* + * 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. + ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(1, 3600)); + UserGroupInformation loginUser = UserGroupInformation.getCurrentUser(); + + try (MockedStatic 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); + } +} From d37d1b35ac9b81236322cded8aa9075023845fa1 Mon Sep 17 00:00:00 2001 From: Henrib Date: Thu, 20 Aug 2026 18:27:48 +0200 Subject: [PATCH 2/2] HIVE-29818: Address review comments on ServletSecurity UGI cache - Rename UgiKey.realUser to effectiveUser (impersonated user, not the Hadoop real user) - Keep cache size as long end-to-end, dropping the truncating int cast - Run removal-listener cleanup on ForkJoinPool.commonPool() instead of the request thread; add a test-only constructor to inject a synchronous executor - Downgrade proxy-user creation log from INFO to DEBUG --- .../hive/metastore/ServletSecurity.java | 25 ++++++++++++++----- .../hive/metastore/TestServletSecurity.java | 4 ++- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java index 13a3d2acb9af..f75dcdfade58 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ServletSecurity.java @@ -31,6 +31,8 @@ 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; @@ -120,7 +122,7 @@ public static AuthType fromString(String type) { * 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 realUser, String loginUser) {} + record UgiKey(String effectiveUser, String loginUser) {} public ServletSecurity(AuthType authType, Configuration conf) { this(authType, conf, null); @@ -128,13 +130,20 @@ public ServletSecurity(AuthType authType, Configuration conf) { public ServletSecurity(AuthType authType, Configuration conf, Function> scopeProvider) { + this(authType, conf, scopeProvider, ForkJoinPool.commonPool()); + } + + @VisibleForTesting + ServletSecurity(AuthType authType, Configuration conf, + Function> 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), - (int) MetastoreConf.getLongVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE)); + MetastoreConf.getLongVar(conf, MetastoreConf.ConfVars.CATALOG_SERVLET_UGI_CACHE_SIZE), + cacheCleanupExecutor); } /** @@ -142,9 +151,13 @@ public ServletSecurity(AuthType authType, Configuration conf, * * @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 createCacheWithConfig(long expirationMs, int maxSize) { + private Cache 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 @@ -167,7 +180,7 @@ private Cache createCacheWithConfig(long expiratio Caffeine builder = Caffeine.newBuilder() .maximumSize(maxSize) - .executor(Runnable::run) + .executor(cacheCleanupExecutor) .removalListener(cleanupListener); if (expirationMs > 0) { @@ -190,8 +203,8 @@ private Cache createCacheWithConfig(long expiratio @VisibleForTesting UserGroupInformation getProxyUser(String userName, UserGroupInformation loginUser) { return proxyUserCache.get(new UgiKey(userName, loginUser.getUserName()), key -> { - LOG.info("Creating proxy user for: {}", key.realUser()); - return UserGroupInformation.createProxyUser(key.realUser(), loginUser); + LOG.debug("Creating proxy user for: {}", key.effectiveUser()); + return UserGroupInformation.createProxyUser(key.effectiveUser(), loginUser); }); } diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java index e91b1907ffbe..e0146f8b6cf1 100644 --- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestServletSecurity.java @@ -73,7 +73,9 @@ public void testDistinctUsersGetDistinctProxies() throws Exception { @Test public void testEvictionClosesFileSystemForUgi() throws Exception { // A cache bounded to a single entry: inserting a second user evicts the first. - ServletSecurity security = new ServletSecurity(ServletSecurity.AuthType.JWT, confWithCache(1, 3600)); + // 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 fsMock = Mockito.mockStatic(FileSystem.class)) {