diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java index d3bcbf42c010..168d7650a866 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java @@ -126,16 +126,19 @@ public boolean run(CompactorContext context) } else { UserGroupInformation ugi = UserGroupInformation.createProxyUser(context.getCompactionInfo().runAs, UserGroupInformation.getLoginUser()); - ugi.doAs((PrivilegedExceptionAction) () -> { - run(context.getConf(), context.getTable(), context.getSd(), - context.getValidWriteIdList(), context.getCompactionInfo(), context.getAcidDirectory()); - return null; - }); try { - FileSystem.closeAllForUGI(ugi); - } catch (IOException exception) { - LOG.error("Could not clean up file-system handles for UGI: " + ugi + " for " + context.getCompactionInfo().getFullPartitionName(), - exception); + ugi.doAs((PrivilegedExceptionAction) () -> { + run(context.getConf(), context.getTable(), context.getSd(), + context.getValidWriteIdList(), context.getCompactionInfo(), context.getAcidDirectory()); + return null; + }); + } finally { + try { + FileSystem.closeAllForUGI(ugi); + } catch (IOException exception) { + LOG.error("Could not clean up file-system handles for UGI: " + ugi + " for " + context.getCompactionInfo().getFullPartitionName(), + exception); + } } } return true; diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java index 80c978fc5696..4bf8704d703e 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java @@ -257,14 +257,17 @@ protected Boolean findNextCompactionAndExecute(boolean collectGenericStats, bool CompactionService finalCompactionService = compactionService; CompactionInfo finalCi = ci; - ugi.doAs((PrivilegedExceptionAction) () -> { - finalCompactionService.cleanupResultDirs(finalCi); - return null; - }); try { - FileSystem.closeAllForUGI(ugi); - } catch (IOException ex) { - LOG.error("Could not clean up file-system handles for UGI: " + ugi, e); + ugi.doAs((PrivilegedExceptionAction) () -> { + finalCompactionService.cleanupResultDirs(finalCi); + return null; + }); + } finally { + try { + FileSystem.closeAllForUGI(ugi); + } catch (IOException ex) { + LOG.error("Could not clean up file-system handles for UGI: " + ugi, e); + } } } } diff --git a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestFileSystemCacheLeakFix.java b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestFileSystemCacheLeakFix.java new file mode 100644 index 000000000000..4af332f9b4e5 --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestFileSystemCacheLeakFix.java @@ -0,0 +1,218 @@ +/* + * 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.ql.txn.compactor; + + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.*; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.hive.common.ValidReaderWriteIdList; +import org.apache.hadoop.hive.common.ValidWriteIdList; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.IMetaStoreClient; +import org.apache.hadoop.hive.metastore.api.CompactionType; +import org.apache.hadoop.hive.metastore.api.StorageDescriptor; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.txn.TxnUtils; +import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo; +import org.apache.hadoop.hive.ql.io.AcidDirectory; +import org.apache.hadoop.security.AccessControlException; +import org.apache.hadoop.util.Progressable; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.net.URI; +import java.util.Map; + +/** + * Regression tests for HIVE-29708: FileSystem.CACHE leak due to missing finally block + * in three compactor locations - TxnUtils.findUserToRunAs(), MRCompactor.run(). + */ +public class TestFileSystemCacheLeakFix { + + private static final String TEST_USER = "testowner"; + private static final int CYCLES = 5; + + @Test + public void testTxnUtils_noFileSystemCacheLeakWhenProxyDoAsFails() throws Exception { + // Verify FileSystem.CACHE does not grow when proxy user also lacks HDFS permission, + // causing doAs() to throw and closeAllForUGI to be skipped in findUserToRunAs(). + HiveConf conf = buildConf(AccessDeniedFileSystem.class, AccessDeniedFileSystem.SCHEME); + Table table = buildTable(AccessDeniedFileSystem.SCHEME); + assertNoCacheGrowth("TxnUtils.findUserToRunAs", + () -> TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf)); + } + + @Test + public void testTxnUtils_noFileSystemCacheLeakWhenServerUserSucceeds() throws Exception { + // Verify FileSystem.CACHE does not grow when server user successfully stats the + // table location - no proxy UGI is created, so no CACHE entry is added. + HiveConf conf = buildConf(AccessGrantedFileSystem.class, AccessGrantedFileSystem.SCHEME); + Table table = buildTable(AccessGrantedFileSystem.SCHEME); + assertNoCacheGrowth("TxnUtils.findUserToRunAs (success path)", () -> { + String owner = TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf); + Assert.assertEquals(AccessGrantedFileSystem.FAKE_OWNER, owner); + }); + } + + @Test + public void testMRCompactor_noFileSystemCacheLeakWhenRunDoAsFails() throws Exception { + // Verify FileSystem.CACHE does not grow when the actual MRCompactor.run(CompactorContext) + // proxy path accesses FileSystem inside doAs() and then throws. + // The inner run() is spied to force FileSystem.get() before throwing, ensuring the + // proxy UGI CACHE entry is created and must be cleaned up via closeAllForUGI. + HiveConf conf = buildConf(AccessDeniedFileSystem.class, AccessDeniedFileSystem.SCHEME); + MRCompactor compactor = Mockito.spy(new MRCompactor(Mockito.mock(IMetaStoreClient.class))); + Mockito.doAnswer((InvocationOnMock inv) -> { + new Path(AccessDeniedFileSystem.SCHEME + "://host/table").getFileSystem(conf); + throw new IOException("simulated MR failure"); + }).when(compactor).run( + Mockito.any(HiveConf.class), + Mockito.any(Table.class), + Mockito.any(StorageDescriptor.class), + Mockito.any(ValidWriteIdList.class), + Mockito.any(CompactionInfo.class), + Mockito.any(AcidDirectory.class)); + + assertNoCacheGrowth("MRCompactor.run", + () -> { try { compactor.run(buildCompactorContext()); } catch (Exception ignored) {} }); + } + + private void assertNoCacheGrowth(String ctx, RunnableWithException op) throws Exception { + try { + op.run(); + } catch (Exception ignored) { + } + int before = getFileSystemCacheSize(); + for (int i = 0; i < TestFileSystemCacheLeakFix.CYCLES; i++) { + try { + op.run(); + } catch (Exception ignored) { + } + } + int after = getFileSystemCacheSize(); + Assert.assertEquals(ctx + ": CACHE grew by " + (after - before) + " (expected 0)", before, after); + } + + private static HiveConf buildConf(Class fsClass, String scheme) { + HiveConf conf = new HiveConf(); + conf.set("fs." + scheme + ".impl", fsClass.getName()); + conf.set("fs." + scheme + ".impl.disable.cache", "false"); + return conf; + } + + private static Table buildTable(String scheme) { + StorageDescriptor sd = new StorageDescriptor(); + sd.setLocation(scheme + ":///warehouse/testdb.db/testtable"); + Table table = new Table(); + table.setDbName("testdb"); + table.setTableName("testtable"); + table.setOwner(TestFileSystemCacheLeakFix.TEST_USER); + table.setSd(sd); + return table; + } + + private static CompactorContext buildCompactorContext() { + HiveConf conf = buildConf(AccessDeniedFileSystem.class, AccessDeniedFileSystem.SCHEME); + StorageDescriptor sd = new StorageDescriptor(); + sd.setLocation(AccessDeniedFileSystem.SCHEME + ":///warehouse/testdb.db/testtable"); + Table table = new Table(); + table.setDbName("testdb"); + table.setTableName("testtable"); + table.setOwner(TestFileSystemCacheLeakFix.TEST_USER); + table.setSd(sd); + CompactionInfo ci = new CompactionInfo("testdb", "testtable", null, CompactionType.MAJOR); + ci.runAs = TestFileSystemCacheLeakFix.TEST_USER; + return new CompactorContext(conf, table, null, sd, + new ValidReaderWriteIdList(), ci, Mockito.mock(AcidDirectory.class)); + } + + static int getFileSystemCacheSize() throws Exception { + Field cacheField = FileSystem.class.getDeclaredField("CACHE"); + cacheField.setAccessible(true); + Object cache = cacheField.get(null); + Field mapField = cache.getClass().getDeclaredField("map"); + mapField.setAccessible(true); + return ((Map) mapField.get(cache)).size(); + } + + @FunctionalInterface + interface RunnableWithException { + void run() throws Exception; + } + + /** Mock FileSystem that always throws AccessControlException on getFileStatus, + * simulating an HDFS directory with drwx------ permissions. */ + public static class AccessDeniedFileSystem extends FileSystem { + public static final String SCHEME = "testcachefix"; + + @Override + public URI getUri() { + return URI.create(SCHEME + ":///"); + } + + @Override + public FileStatus getFileStatus(Path f) throws IOException { + throw new AccessControlException( + "Permission denied: access=EXECUTE, inode=\"" + f + "\":hdfs:hdfs:drwx------"); + } + + @Override public void initialize(URI uri, Configuration conf) throws IOException {} + @Override public FSDataInputStream open(Path f, int bufferSize) throws IOException { return null; } + @Override public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { return null; } + @Override public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException { return null; } + @Override public boolean rename(Path src, Path dst) throws IOException { return false; } + @Override public boolean delete(Path f, boolean recursive) throws IOException { return false; } + @Override public FileStatus[] listStatus(Path f) throws IOException { return new FileStatus[0]; } + @Override public void setWorkingDirectory(Path newDir) {} + @Override public Path getWorkingDirectory() { return new Path(SCHEME + ":///"); } + @Override public boolean mkdirs(Path f, FsPermission permission) throws IOException { return false; } + } + + /** Mock FileSystem that always succeeds on getFileStatus, returning a fixed owner. */ + public static class AccessGrantedFileSystem extends FileSystem { + public static final String SCHEME = "testsuccess"; + public static final String FAKE_OWNER = "tableowner"; + + @Override + public URI getUri() { + return URI.create(SCHEME + ":///"); + } + + @Override + public FileStatus getFileStatus(Path f) throws IOException { + return new FileStatus(0, true, 0, 0, 0, 0, FsPermission.getDirDefault(), FAKE_OWNER, FAKE_OWNER, f); + } + + @Override public void initialize(URI uri, Configuration conf) throws IOException {} + @Override public FSDataInputStream open(Path f, int bufferSize) throws IOException { return null; } + @Override public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { return null; } + @Override public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException { return null; } + @Override public boolean rename(Path src, Path dst) throws IOException { return false; } + @Override public boolean delete(Path f, boolean recursive) throws IOException { return false; } + @Override public FileStatus[] listStatus(Path f) throws IOException { return new FileStatus[0]; } + @Override public void setWorkingDirectory(Path newDir) {} + @Override public Path getWorkingDirectory() { return new Path(SCHEME + ":///"); } + @Override public boolean mkdirs(Path f, FsPermission permission) throws IOException { return false; } + } +} \ No newline at end of file diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java index 5c5ca7c7bfad..79d6900222b6 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java @@ -572,18 +572,20 @@ public static String findUserToRunAs(String location, Table t, Configuration con UserGroupInformation ugi = UserGroupInformation.createProxyUser(t.getOwner(), UserGroupInformation.getLoginUser()); - ugi.doAs((PrivilegedExceptionAction) () -> { - // need to use a new filesystem object here to have the correct ugi - FileSystem proxyFs = p.getFileSystem(conf); - FileStatus stat = proxyFs.getFileStatus(p); - wrapper.add(stat.getOwner()); - return null; - }); - try { - FileSystem.closeAllForUGI(ugi); - } catch (IOException exception) { - LOG.error("Could not clean up file-system handles for UGI: " + ugi, exception); + ugi.doAs((PrivilegedExceptionAction) () -> { + // need to use a new filesystem object here to have the correct ugi + FileSystem proxyFs = p.getFileSystem(conf); + FileStatus stat = proxyFs.getFileStatus(p); + wrapper.add(stat.getOwner()); + return null; + }); + } finally { + try { + FileSystem.closeAllForUGI(ugi); + } catch (IOException exception) { + LOG.error("Could not clean up file-system handles for UGI: " + ugi, exception); + } } if (wrapper.size() == 1) { LOG.debug("Running job as {}", wrapper.get(0));