diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java index 7e039e74f03..7d511994810 100644 --- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java +++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java @@ -313,6 +313,10 @@ public enum Property { + " user-implementations of pluggable Accumulo features, such as the balancer" + " or volume chooser.", "2.0.0"), + GENERAL_BULK_SOURCE_DIRS("general.bulk.user.allowed.source.dir", "{}", PropertyType.JSON, + "Map of Accumulo user name to Java regular expression, where the regular expression " + + "will specify the allowed source location of bulk import files.", + "2.1.7"), GENERAL_CACHE_MANAGER_IMPL("general.block.cache.manager.class", "org.apache.accumulo.core.file.blockfile.cache.lru.LruBlockCacheManager", PropertyType.STRING, "Specifies the class name of the block cache factory implementation." @@ -2126,6 +2130,7 @@ public static boolean isValidZooPropertyKey(String key) { || key.startsWith(Property.MASTER_PREFIX.getKey()) || key.startsWith(Property.GC_PREFIX.getKey()) || key.startsWith(Property.GENERAL_ARBITRARY_PROP_PREFIX.getKey()) + || key.equals(Property.GENERAL_BULK_SOURCE_DIRS.getKey()) || key.equals(Property.GENERAL_FILE_NAME_ALLOCATION_BATCH_SIZE_MIN.getKey()) || key.equals(Property.GENERAL_FILE_NAME_ALLOCATION_BATCH_SIZE_MAX.getKey()) || key.equals(Property.GENERAL_SERVER_ITERATOR_OPTIONS_COMPRESSION_ALGO.getKey()) diff --git a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java b/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java index 45e665c8823..d808c9fa6a1 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java +++ b/server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java @@ -20,10 +20,13 @@ import static java.nio.charset.StandardCharsets.UTF_8; +import java.lang.reflect.Type; import java.nio.ByteBuffer; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.function.Supplier; +import java.util.regex.Pattern; import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.client.AccumuloSecurityException; @@ -59,6 +62,10 @@ import org.slf4j.LoggerFactory; import com.google.common.base.Suppliers; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonSyntaxException; +import com.google.gson.reflect.TypeToken; /** * Utility class for performing various security operations with the appropriate checks @@ -73,6 +80,8 @@ public class SecurityOperation { private final Supplier rootUserName; private final ZooCache zooCache; private final String zkUserPath; + private final Gson gson = new GsonBuilder().create(); + private final Type jsonMapType = new TypeToken>() {}.getType(); protected final ServerContext context; @@ -537,7 +546,33 @@ protected boolean canDeleteRange(TCredentials c, TableId tableId, String tableNa protected boolean canBulkImport(TCredentials c, TableId tableId, String tableName, String dir, String failDir, NamespaceId namespaceId) throws ThriftSecurityException { authenticate(c); - return hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false); + final boolean tablePerm = + hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false); + + final String user = c.getPrincipal(); + final String allowedLocations = + context.getConfiguration().get(Property.GENERAL_BULK_SOURCE_DIRS); + boolean locationAllowed = false; + try { + final Map allowedLocationsMap = gson.fromJson(allowedLocations, jsonMapType); + final String userSourceLocationPattern = allowedLocationsMap.get(user); + if (userSourceLocationPattern == null) { + log.error("User " + user + " does not have an allowed source dir location in property " + + Property.GENERAL_BULK_SOURCE_DIRS.getKey()); + } else if (Pattern.matches(userSourceLocationPattern, dir)) { + locationAllowed = true; + } + } catch (JsonSyntaxException e) { + log.error( + "Property " + Property.GENERAL_BULK_SOURCE_DIRS.getKey() + " value is not valid json", e); + } + + if (!locationAllowed) { + log.error( + "Bulk Import into table {} by user {} not allowed from directory {}, check table property {} value: {}", + tableId, user, dir, Property.GENERAL_BULK_SOURCE_DIRS.getKey(), allowedLocations); + } + return tablePerm && locationAllowed; } protected boolean canCompact(TCredentials c, TableId tableId, NamespaceId namespaceId) diff --git a/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java b/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java index 8f3940e2134..1a74c4f3a48 100644 --- a/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java +++ b/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java @@ -56,6 +56,7 @@ import org.apache.accumulo.core.security.TablePermission; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; import org.apache.accumulo.server.security.AuditedSecurityOperation; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.accumulo.test.functional.ConfigurableMacBase; import org.apache.commons.io.FileUtils; import org.apache.hadoop.io.Text; @@ -350,6 +351,9 @@ public void testImportExportOperationsAudits() throws AccumuloSecurityException, Collections.singleton(exportDir.toString()), ImportConfiguration.empty()); // Now do a Directory (bulk) import of the same data. + BulkIT.setBulkSourceLocations(auditAccumuloClient, getCluster().getFileSystem(), + exportDirBulk.getParentFile()); + auditAccumuloClient.tableOperations().create(THIRD_TEST_TABLE_NAME); auditAccumuloClient.tableOperations().importDirectory(exportDirBulk.toString()) .to(THIRD_TEST_TABLE_NAME).load(); diff --git a/test/src/main/java/org/apache/accumulo/test/BulkImportMonitoringIT.java b/test/src/main/java/org/apache/accumulo/test/BulkImportMonitoringIT.java index f8d423fa84c..68c4bb9f010 100644 --- a/test/src/main/java/org/apache/accumulo/test/BulkImportMonitoringIT.java +++ b/test/src/main/java/org/apache/accumulo/test/BulkImportMonitoringIT.java @@ -46,6 +46,7 @@ import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.minicluster.ServerType; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.accumulo.test.functional.ConfigurableMacBase; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; @@ -95,6 +96,8 @@ public void test() throws Exception { fs.delete(base, true); fs.mkdirs(base); + BulkIT.setBulkSourceLocations(c, fs, base); + ExecutorService es = Executors.newFixedThreadPool(5); List>> futures = new ArrayList<>(); for (int i = 0; i < 10; i++) { diff --git a/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java b/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java index 0a70abf10ec..d09a1dc7e4b 100644 --- a/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java +++ b/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java @@ -30,6 +30,7 @@ import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FsShell; @@ -103,6 +104,7 @@ public void testBulkImportFailure() throws Exception { // because // of the balancer moving tablets around // and then we get to verify that the bug is actually fixed. + BulkIT.setBulkSourceLocations(client, fs, bulk.getParent()); to.importDirectory(bulk.toString()).to(tableName).load(); // The bug is that some tablets don't get imported into. diff --git a/test/src/main/java/org/apache/accumulo/test/BulkImportVolumeIT.java b/test/src/main/java/org/apache/accumulo/test/BulkImportVolumeIT.java index 759b3f6e57b..b62b0e39a70 100644 --- a/test/src/main/java/org/apache/accumulo/test/BulkImportVolumeIT.java +++ b/test/src/main/java/org/apache/accumulo/test/BulkImportVolumeIT.java @@ -29,6 +29,7 @@ import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -96,6 +97,7 @@ public void testBulkImportFailure() throws Exception { log.info("bogus: {}", bogus); assertTrue(fs.exists(bogus)); log.info("Importing {} into {} with failures directory {}", bulk, tableName, err); + BulkIT.setBulkSourceLocations(client, fs, bulk.getParent()); client.tableOperations().importDirectory(tableName, bulk.toString(), err.toString(), false); assertEquals(1, fs.listStatus(err).length); } diff --git a/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java b/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java index c925e089c54..bcb27639a7b 100644 --- a/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java +++ b/test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java @@ -47,6 +47,7 @@ import org.apache.accumulo.core.spi.crypto.NoCryptoServiceFactory; import org.apache.accumulo.minicluster.ServerType; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.accumulo.test.functional.ConfigurableMacBase; import org.apache.accumulo.test.functional.FunctionalTestUtils; import org.apache.hadoop.conf.Configuration; @@ -122,6 +123,7 @@ public void compareOldNewBulkImportTest() throws Exception { final Path base = new Path(basePath, "testBulkLoad" + tableName); fs.delete(base, true); fs.mkdirs(base); + BulkIT.setBulkSourceLocations(c, fs, base); ExecutorService es = Executors.newFixedThreadPool(5); List> futures = new ArrayList<>(); diff --git a/test/src/main/java/org/apache/accumulo/test/ScanConsistencyIT.java b/test/src/main/java/org/apache/accumulo/test/ScanConsistencyIT.java index c75f5c413f3..82fbf5316cd 100644 --- a/test/src/main/java/org/apache/accumulo/test/ScanConsistencyIT.java +++ b/test/src/main/java/org/apache/accumulo/test/ScanConsistencyIT.java @@ -58,6 +58,7 @@ import org.apache.accumulo.core.iterators.IteratorEnvironment; import org.apache.accumulo.core.iterators.SortedKeyValueIterator; import org.apache.accumulo.harness.AccumuloClusterHarness; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -109,6 +110,7 @@ public void testConcurrentScanConsistency() throws Exception { TestContext testContext = new TestContext(client, table, getCluster().getFileSystem(), getCluster().getTemporaryPath().toString()); + BulkIT.setBulkSourceLocations(client, getFileSystem(), getCluster().getTemporaryPath()); List> writeTasks = new ArrayList<>(); List> scanTasks = new ArrayList<>(); diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java index 42477aad9c0..3e16f9a29bf 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java +++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java @@ -30,6 +30,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.File; import java.time.Duration; import java.util.ArrayList; import java.util.EnumSet; @@ -61,6 +62,7 @@ import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.minicluster.ServerType; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.accumulo.test.functional.SlowIterator; import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory; import org.apache.accumulo.test.metrics.TestStatsDSink; @@ -378,6 +380,7 @@ public void testProgressWithBulkImport() throws Exception { String dir = getDir(client, tableName1); log.info("Bulk importing files in dir " + dir + " to table " + tableName2); + BulkIT.setBulkSourceLocations(client, getFileSystem(), new File(dir).getParentFile()); client.tableOperations().importDirectory(dir).to(tableName2).load(); log.info("Finished bulk import"); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkFailureIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkFailureIT.java index 98f3e47f290..492ea3e783d 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkFailureIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkFailureIT.java @@ -131,6 +131,7 @@ protected void runTest(String table, long fateTxid, Loader loader) throws IOExce // determine the files new name and path FileStatus status = fs.listStatus(new Path(bulkDir))[0]; Path bulkLoadPath = fs.makeQualified(status.getPath()); + BulkIT.setBulkSourceLocations(c, fs, bulkLoadPath); // Directly ask the tablet to load the file. loader.load(fateTxid, asCtx, extent, bulkLoadPath, status.getLen(), false); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java index ae67f35d486..c50541e9bb8 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java @@ -19,24 +19,35 @@ package org.apache.accumulo.test.functional; import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.File; import java.io.IOException; import java.time.Duration; +import java.util.HashMap; +import java.util.Map; import org.apache.accumulo.core.client.Accumulo; import org.apache.accumulo.core.client.AccumuloClient; import org.apache.accumulo.core.client.AccumuloException; import org.apache.accumulo.core.client.AccumuloSecurityException; import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.harness.AccumuloClusterHarness; import org.apache.accumulo.test.TestIngest; import org.apache.accumulo.test.TestIngest.IngestParams; import org.apache.accumulo.test.VerifyIngest; import org.apache.accumulo.test.VerifyIngest.VerifyParams; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.gson.GsonBuilder; /** * Tests Old and New Bulk import @@ -44,6 +55,8 @@ @Tag(SUNNY_DAY) public class BulkIT extends AccumuloClusterHarness { + private static final Logger LOG = LoggerFactory.getLogger(BulkIT.class); + private static final int N = 100000; private static final int COUNT = 5; @@ -52,19 +65,68 @@ protected Duration defaultTimeout() { return Duration.ofMinutes(4); } + public static void setBulkSourceLocations(AccumuloClient client, FileSystem fs, + File localDirectory) + throws AccumuloException, AccumuloSecurityException, IllegalArgumentException, IOException { + Path p = FileSystem.get(localDirectory.toURI(), new Configuration()) + .makeQualified(new Path(localDirectory.toString())); + setBulkSourceLocations(client, fs, p); + } + + public static void setBulkSourceLocations(AccumuloClient client, FileSystem fs, Path basePath) + throws AccumuloException, AccumuloSecurityException { + Map userPatterns = new HashMap<>(); + userPatterns.put(client.whoami(), fs.makeQualified(basePath).toString() + "/.*"); + String propVal = new GsonBuilder().create().toJson(userPatterns); + LOG.info("Property set to {}", propVal); + client.instanceOperations().setProperty(Property.GENERAL_BULK_SOURCE_DIRS.getKey(), propVal); + } + @Test public void test() throws Exception { try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) { - runTest(client, getCluster().getFileSystem(), getCluster().getTemporaryPath(), - getUniqueNames(1)[0], this.getClass().getName(), testName(), false); + Path basePath = getCluster().getTemporaryPath(); + FileSystem fs = getCluster().getFileSystem(); + setBulkSourceLocations(client, fs, basePath); + runTest(client, fs, basePath, getUniqueNames(1)[0], this.getClass().getName(), testName(), + false); } } @Test public void testOld() throws Exception { try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) { - runTest(client, getCluster().getFileSystem(), getCluster().getTemporaryPath(), - getUniqueNames(1)[0], this.getClass().getName(), testName(), true); + Path basePath = getCluster().getTemporaryPath(); + FileSystem fs = getCluster().getFileSystem(); + setBulkSourceLocations(client, fs, basePath); + runTest(client, fs, basePath, getUniqueNames(1)[0], this.getClass().getName(), testName(), + true); + } + } + + @Test + public void testWithoutSourceDirProperty() throws Exception { + try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) { + Path basePath = getCluster().getTemporaryPath(); + FileSystem fs = getCluster().getFileSystem(); + AccumuloSecurityException ase = + assertThrows(AccumuloSecurityException.class, () -> runTest(client, fs, basePath, + getUniqueNames(1)[0], this.getClass().getName(), testName(), false)); + assertTrue(ase.getMessage().startsWith("Error PERMISSION_DENIED for user root on table ")); + assertTrue(ase.getMessage().endsWith("User does not have permission to perform this action")); + } + } + + @Test + public void testOldWithoutSourceDirProperty() throws Exception { + try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) { + Path basePath = getCluster().getTemporaryPath(); + FileSystem fs = getCluster().getFileSystem(); + AccumuloSecurityException ase = + assertThrows(AccumuloSecurityException.class, () -> runTest(client, fs, basePath, + getUniqueNames(1)[0], this.getClass().getName(), testName(), true)); + assertTrue(ase.getMessage().startsWith("Error PERMISSION_DENIED for user root on table ")); + assertTrue(ase.getMessage().endsWith("User does not have permission to perform this action")); } } diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java index 458ec594c84..2665b315692 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewIT.java @@ -164,6 +164,8 @@ public void setupBulkTest() throws Exception { aconf = getCluster().getServerContext().getConfiguration(); fs = getCluster().getFileSystem(); rootPath = getCluster().getTemporaryPath().toString(); + BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath()); + } } diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewMetadataSkipIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewMetadataSkipIT.java index 5a29a1ed33b..02057829ed8 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkNewMetadataSkipIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkNewMetadataSkipIT.java @@ -105,7 +105,10 @@ public static String writeNonContiguousData(FileSystem fs, String file, public void setupCluster() throws Exception { super.setupCluster(); // prime the zk connection - try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {} + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + BulkIT.setBulkSourceLocations(c, getCluster().getFileSystem(), + getCluster().getTemporaryPath()); + } } @ParameterizedTest diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkOldIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkOldIT.java index d8df93be07b..709dee46f15 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkOldIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkOldIT.java @@ -83,6 +83,7 @@ public void testBulkFile() throws Exception { Configuration conf = new Configuration(); AccumuloConfiguration aconf = getCluster().getServerContext().getConfiguration(); FileSystem fs = getCluster().getFileSystem(); + BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath()); String rootPath = cluster.getTemporaryPath().toString(); String dir = rootPath + "/bulk_test_diff_files_89723987592_" + getUniqueNames(1)[0]; @@ -128,6 +129,7 @@ public void testExceptionInMetadataUpdate() throws Exception { Configuration conf = new Configuration(); AccumuloConfiguration aconf = getCluster().getServerContext().getConfiguration(); FileSystem fs = getCluster().getFileSystem(); + BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath()); String rootPath = cluster.getTemporaryPath().toString(); String dir = rootPath + "/bulk_test_diff_files_89723987592_" + getUniqueNames(1)[0]; diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java index 9fd60d0dfb3..9c8f0bba837 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java @@ -95,6 +95,7 @@ public void testBulkSplitOptimization() throws Exception { FileSystem fs = cluster.getFileSystem(); Path testDir = new Path(cluster.getTemporaryPath(), "testmf"); fs.deleteOnExit(testDir); + BulkIT.setBulkSourceLocations(c, fs, cluster.getTemporaryPath()); FunctionalTestUtils.createRFiles(c, fs, testDir.toString(), ROWS, SPLITS, 8); FileStatus[] stats = fs.listStatus(testDir); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java index a927bc4a3cc..503c6eed3da 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java @@ -626,6 +626,7 @@ public void testSuccessfulCompaction() throws Exception { fs.deleteOnExit(testrf); FunctionalTestUtils.createRFiles(c, fs, testrf.toString(), 500000, 59, 4); + BulkIT.setBulkSourceLocations(c, fs, root); c.tableOperations().importDirectory(testrf.toString()).to(tableName).load(); int beforeCount = countFiles(c, tableName); diff --git a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java index 18db2d9c19b..0da839bd2e0 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/SplitIT.java @@ -247,6 +247,7 @@ public void bulkImportThatCantSplitHangsCompaction() throws Exception { Random random = new Random(); byte[] val = new byte[100]; + BulkIT.setBulkSourceLocations(c, getFileSystem(), getCluster().getTemporaryPath()); String dir = getDir(); String file = dir + "/f1.rf"; diff --git a/test/src/main/java/org/apache/accumulo/test/functional/SslIT.java b/test/src/main/java/org/apache/accumulo/test/functional/SslIT.java index 008237e3170..c20377d9b79 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/SslIT.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/SslIT.java @@ -81,6 +81,7 @@ public void adminStop() throws Exception { public void bulk() throws Exception { Properties props = getClientProperties(); try (AccumuloClient client = Accumulo.newClient().from(props).build()) { + BulkIT.setBulkSourceLocations(client, cluster.getFileSystem(), cluster.getTemporaryPath()); BulkIT.runTest(client, cluster.getFileSystem(), new Path(getCluster().getConfig().getDir().getAbsolutePath(), "tmp"), getUniqueNames(1)[0], this.getClass().getName(), testName(), true); diff --git a/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java b/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java index 00e7cfdc033..5afe9fb7294 100644 --- a/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java +++ b/test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java @@ -81,6 +81,7 @@ import org.apache.accumulo.harness.SharedMiniClusterBase; import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; import org.apache.accumulo.test.compaction.TestCompactionStrategy; +import org.apache.accumulo.test.functional.BulkIT; import org.apache.accumulo.test.functional.SlowIterator; import org.apache.accumulo.test.util.Wait; import org.apache.hadoop.conf.Configuration; @@ -1343,6 +1344,10 @@ public void importDirectoryOld() throws Exception { String nonce = generateNonce(); File importDir = createRFiles(conf, fs, table, nonce); + try (AccumuloClient client = + Accumulo.newClient().from(getCluster().getClientProperties()).build()) { + BulkIT.setBulkSourceLocations(client, fs, importDir.getParentFile()); + } ts.exec("createtable " + table, true); ts.exec("importdirectory " + importDir + " " + errorsDir + " true", true); ts.exec("scan -r 00000000", true, "0-->" + nonce, true); @@ -1357,6 +1362,10 @@ public void importDirectory() throws Exception { String nonce = generateNonce(); FileSystem fs = FileSystem.get(conf); File importDir = createRFiles(conf, fs, table, nonce); + try (AccumuloClient client = + Accumulo.newClient().from(getCluster().getClientProperties()).build()) { + BulkIT.setBulkSourceLocations(client, fs, importDir.getParentFile()); + } ts.exec("createtable " + table, true); ts.exec("importdirectory " + importDir + " true", true); ts.exec("scan -r 00000000", true, "0-->" + nonce, true); @@ -1373,6 +1382,10 @@ public void importDirectoryWithOptions() throws Exception { String nonce = generateNonce(); FileSystem fs = FileSystem.get(conf); File importDir = createRFiles(conf, fs, table, nonce); + try (AccumuloClient client = + Accumulo.newClient().from(getCluster().getClientProperties()).build()) { + BulkIT.setBulkSourceLocations(client, fs, importDir.getParentFile()); + } ts.exec("createtable " + table, true); ts.exec("notable", true); ts.exec("importdirectory -t " + table + " -i " + importDir + " true", true); @@ -1951,6 +1964,12 @@ public void importDirectoryCmdFmt() throws Exception { File errorsDir = new File(rootPath, "errors_" + table); assertTrue(errorsDir.mkdir()); + try (AccumuloClient client = + Accumulo.newClient().from(getCluster().getClientProperties()).build()) { + BulkIT.setBulkSourceLocations(client, getCluster().getFileSystem(), + importDir.getParentFile()); + } + // expect fail - table does not exist. ts.exec(String.format("importdirectory -t %s %s %s false", table, importDir, errorsDir), false, "TableNotFoundException");