diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java index 08a4506..00d3d0d 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java @@ -3757,10 +3757,13 @@ private static Path mvFile(HiveConf conf, FileSystem sourceFs, Path sourcePath, } else if (isSrcLocal) { destFs.copyFromLocalFile(sourcePath, destFilePath); } else { - FileUtils.copy(sourceFs, sourcePath, destFs, destFilePath, + if (!FileUtils.copy(sourceFs, sourcePath, destFs, destFilePath, true, // delete source false, // overwrite destination - conf); + conf)) { + LOG.error("Copy failed for source: " + sourcePath + " to destination: " + destFilePath); + throw new IOException("File copy failed."); + } } return destFilePath; } diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/CopyUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/CopyUtils.java index 75dcaa3..461f558 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/CopyUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/CopyUtils.java @@ -262,15 +262,17 @@ private void doDistCpCopyOnce(FileSystem sourceFs, List srcList, Path dest RAW_RESERVED_VIRTUAL_PATH + destinationUri.getPath()); } - FileUtils.distCp( + if (!FileUtils.distCp( sourceFs, // source file system srcList, // list of source paths destination, false, usePrivilegedUser ? copyAsUser : null, hiveConf, - ShimLoader.getHadoopShims() - ); + ShimLoader.getHadoopShims())) { + LOG.error("Distcp failed to copy files: " + srcList + " to destination: " + destination); + throw new IOException("Distcp operation failed."); + } } private void doRegularCopyOnce(FileSystem sourceFs, List srcList, FileSystem destinationFs, @@ -319,7 +321,7 @@ public void doCopy(Path destination, List srcPaths) throws IOException, Lo 3. aggregate fileSize of all source Paths(can be directory / file) is less than configured size. 4. number of files of all source Paths(can be directory / file) is less than configured size. */ - private boolean regularCopy(FileSystem destinationFs, FileSystem sourceFs, List fileList) + boolean regularCopy(FileSystem destinationFs, FileSystem sourceFs, List fileList) throws IOException { if (hiveInTest) { return true; diff --git a/ql/src/test/org/apache/hadoop/hive/ql/parse/repl/TestCopyUtils.java b/ql/src/test/org/apache/hadoop/hive/ql/parse/repl/TestCopyUtils.java index 8714660..7bd660b 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/parse/repl/TestCopyUtils.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/parse/repl/TestCopyUtils.java @@ -18,30 +18,89 @@ package org.apache.hadoop.hive.ql.parse.repl; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.common.FileUtils; import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.ReplChangeManager; +import org.apache.hadoop.hive.shims.ShimLoader; +import org.apache.hadoop.hive.shims.Utils; +import org.apache.hadoop.security.UserGroupInformation; import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; import static org.junit.Assert.assertFalse; +import static org.mockito.Matchers.anyBoolean; +import static org.mockito.Matchers.anyListOf; +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.same; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.powermock.api.mockito.PowerMockito.mockStatic; +import static org.powermock.api.mockito.PowerMockito.when; +/** + * Unit Test class for CopyUtils class. + */ +@RunWith(PowerMockRunner.class) +@PrepareForTest({ CopyUtils.class, FileUtils.class, Utils.class, UserGroupInformation.class}) +@PowerMockIgnore({ "javax.management.*" }) public class TestCopyUtils { /* Distcp currently does not copy a single file in a distributed manner hence we dont care about the size of file, if there is only file, we dont want to launch distcp. */ @Test - public void distcpShouldNotBeCalledOnlyForOneFile() { - HiveConf conf = new HiveConf(); - conf.setLongVar(HiveConf.ConfVars.HIVE_EXEC_COPYFILE_MAXSIZE, 1); + public void distcpShouldNotBeCalledOnlyForOneFile() throws Exception { + mockStatic(UserGroupInformation.class); + when(UserGroupInformation.getCurrentUser()).thenReturn(mock(UserGroupInformation.class)); + + HiveConf conf = Mockito.spy(new HiveConf()); + doReturn(1L).when(conf).getLong(HiveConf.ConfVars.HIVE_EXEC_COPYFILE_MAXSIZE.varname, 32L * 1024 * 1024); CopyUtils copyUtils = new CopyUtils("", conf); long MB_128 = 128 * 1024 * 1024; assertFalse(copyUtils.limitReachedForLocalCopy(MB_128, 1L)); } @Test - public void distcpShouldNotBeCalledForSmallerFileSize() { - HiveConf conf = new HiveConf(); + public void distcpShouldNotBeCalledForSmallerFileSize() throws Exception { + mockStatic(UserGroupInformation.class); + when(UserGroupInformation.getCurrentUser()).thenReturn(mock(UserGroupInformation.class)); + + HiveConf conf = Mockito.spy(new HiveConf()); CopyUtils copyUtils = new CopyUtils("", conf); long MB_16 = 16 * 1024 * 1024; assertFalse(copyUtils.limitReachedForLocalCopy(MB_16, 100L)); } + + @Test(expected = IOException.class) + public void shouldThrowExceptionOnDistcpFailure() throws Exception { + Path destination = mock(Path.class); + Path source = mock(Path.class); + FileSystem fs = mock(FileSystem.class); + List srcPaths = Arrays.asList(source, source); + HiveConf conf = mock(HiveConf.class); + CopyUtils copyUtils = Mockito.spy(new CopyUtils(null, conf)); + + mockStatic(FileUtils.class); + mockStatic(Utils.class); + when(destination.getFileSystem(same(conf))).thenReturn(fs); + when(source.getFileSystem(same(conf))).thenReturn(fs); + when(FileUtils.distCp(same(fs), anyListOf(Path.class), same(destination), + anyBoolean(), eq(null), same(conf), + same(ShimLoader.getHadoopShims()))) + .thenReturn(false); + when(Utils.getUGI()).thenReturn(mock(UserGroupInformation.class)); + doReturn(false).when(copyUtils).regularCopy(same(fs), same(fs), anyListOf(ReplChangeManager.FileInfo.class)); + + copyUtils.doCopy(destination, srcPaths); + } } \ No newline at end of file