From 519ab34eb5a612692b6db13849d38aa454999b17 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Thu, 24 Sep 2026 18:19:22 +0000 Subject: [PATCH 1/6] wip --- .../apache/accumulo/core/conf/Property.java | 5 +++++ .../server/security/SecurityOperation.java | 22 ++++++++++++++++++- .../accumulo/test/functional/BulkIT.java | 19 ++++++++++++++++ 3 files changed, 45 insertions(+), 1 deletion(-) 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..3c0784b454d 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 @@ -1305,6 +1305,11 @@ public enum Property { + " this feature. A non-zero value enables this feature and the Manager will setup a new scanner" + " when the tablet metadata distance is above the supplied value.", "2.1.4"), + TABLE_BULK_SOURCE_DIRS("table.bulk.allowed.source.dir.patterns", + "^((?!\\/accumulo\\/tables\\/).)*$", PropertyType.STRING, + "A comma-separated list of regular expressions where source bulk import files for the table may located." + + " The default value prevents using rfiles from an Accumulo table directory from being the source directory.", + "2.1.7"), TABLE_DURABILITY("table.durability", "sync", PropertyType.DURABILITY, "The durability of writes to tables includes ensuring that mutations written" + " by clients are persisted in the write-ahead log and that files written" 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..fad632222ea 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 @@ -24,6 +24,7 @@ import java.util.List; 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; @@ -537,7 +538,26 @@ 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); + boolean locationMatch = false; + String allowedLocations = + context.getTableConfiguration(tableId).get(Property.TABLE_BULK_SOURCE_DIRS); + if (allowedLocations.isBlank()) { + locationMatch = true; + } else { + for (String location : allowedLocations.split(",")) { + if (Pattern.matches(location, dir)) { + locationMatch = true; + break; + } + } + } + if (!locationMatch) { + log.error("Bulk Import into table {} not allowed from directory {}, check table property {}", + tableId, dir, Property.TABLE_BULK_SOURCE_DIRS.getKey()); + } + return tablePerm && locationMatch; } protected boolean canCompact(TCredentials c, TableId tableId, NamespaceId namespaceId) 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..2c084d5c5d9 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,6 +19,7 @@ package org.apache.accumulo.test.functional; import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.IOException; import java.time.Duration; @@ -34,7 +35,9 @@ import org.apache.accumulo.test.VerifyIngest; import org.apache.accumulo.test.VerifyIngest.VerifyParams; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.LocatedFileStatus; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.RemoteIterator; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; @@ -108,6 +111,22 @@ static void runTest(AccumuloClient c, FileSystem fs, Path basePath, String table verifyParams.startRow = N; verifyParams.rows = 1; VerifyIngest.verifyIngest(c, verifyParams); + + // Create a subdirectory that is not allowed and try to bulk import from + // that location + Path a = new Path(base, "accumulo"); + fs.deleteOnExit(a); + Path t = new Path(base, "tables"); + fs.deleteOnExit(t); + fs.mkdirs(t); + params = new IngestParams(c.properties(), tableName, N); + params.outputFile = new Path(t, String.format(fileFormat, N)).toString(); + params.startRow = N; + params.rows = 1; + // create an rfile with one entry, there was a bug with this: + TestIngest.ingest(c, fs, params); + bulkLoad(c, tableName, bulkFailures, t, useOld); + } @SuppressWarnings("deprecation") From ecc06f67b221109e53d998b8ddd5a09546670502 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Thu, 24 Sep 2026 21:50:41 +0000 Subject: [PATCH 2/6] Added property to restrict bulk import source file locations --- .../apache/accumulo/core/conf/Property.java | 11 +++---- .../server/security/SecurityOperation.java | 30 ++++++++----------- .../accumulo/test/functional/BulkIT.java | 12 ++++---- 3 files changed, 25 insertions(+), 28 deletions(-) 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 3c0784b454d..690550faf13 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,12 @@ 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.allowed.source.dir.pattern", "^(?!.*/accumulo/tables).*$", + PropertyType.STRING, + "A regular expressions where source bulk import files for the table may located." + + " The default value prevents using rfiles from the default Accumulo table directory" + + " path from being the source directory.", + "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." @@ -1305,11 +1311,6 @@ public enum Property { + " this feature. A non-zero value enables this feature and the Manager will setup a new scanner" + " when the tablet metadata distance is above the supplied value.", "2.1.4"), - TABLE_BULK_SOURCE_DIRS("table.bulk.allowed.source.dir.patterns", - "^((?!\\/accumulo\\/tables\\/).)*$", PropertyType.STRING, - "A comma-separated list of regular expressions where source bulk import files for the table may located." - + " The default value prevents using rfiles from an Accumulo table directory from being the source directory.", - "2.1.7"), TABLE_DURABILITY("table.durability", "sync", PropertyType.DURABILITY, "The durability of writes to tables includes ensuring that mutations written" + " by clients are persisted in the write-ahead log and that files written" 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 fad632222ea..caba2b94286 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 @@ -540,24 +540,18 @@ protected boolean canBulkImport(TCredentials c, TableId tableId, String tableNam authenticate(c); final boolean tablePerm = hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false); - boolean locationMatch = false; - String allowedLocations = - context.getTableConfiguration(tableId).get(Property.TABLE_BULK_SOURCE_DIRS); - if (allowedLocations.isBlank()) { - locationMatch = true; - } else { - for (String location : allowedLocations.split(",")) { - if (Pattern.matches(location, dir)) { - locationMatch = true; - break; - } - } - } - if (!locationMatch) { - log.error("Bulk Import into table {} not allowed from directory {}, check table property {}", - tableId, dir, Property.TABLE_BULK_SOURCE_DIRS.getKey()); - } - return tablePerm && locationMatch; + final String allowedLocations = + context.getTableConfiguration(tableId).get(Property.GENERAL_BULK_SOURCE_DIRS); + boolean locationAllowed = false; + if (allowedLocations.isBlank() || Pattern.matches(allowedLocations, dir)) { + locationAllowed = true; + } + if (!locationAllowed) { + log.error( + "Bulk Import into table {} not allowed from directory {}, check table property {} value: {}", + tableId, 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/functional/BulkIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java index 2c084d5c5d9..67c5539b5f5 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,6 +19,7 @@ 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.IOException; @@ -35,9 +36,7 @@ import org.apache.accumulo.test.VerifyIngest; import org.apache.accumulo.test.VerifyIngest.VerifyParams; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.LocatedFileStatus; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.RemoteIterator; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; @@ -116,16 +115,19 @@ static void runTest(AccumuloClient c, FileSystem fs, Path basePath, String table // that location Path a = new Path(base, "accumulo"); fs.deleteOnExit(a); - Path t = new Path(base, "tables"); + Path t = new Path(a, "tables"); fs.deleteOnExit(t); fs.mkdirs(t); - params = new IngestParams(c.properties(), tableName, N); + params = new IngestParams(c.properties(), tableName, N); params.outputFile = new Path(t, String.format(fileFormat, N)).toString(); params.startRow = N; params.rows = 1; // create an rfile with one entry, there was a bug with this: TestIngest.ingest(c, fs, params); - bulkLoad(c, tableName, bulkFailures, t, useOld); + AccumuloSecurityException ase = assertThrows(AccumuloSecurityException.class, + () -> bulkLoad(c, tableName, bulkFailures, t, useOld)); + assertTrue( + ase.getMessage().contains("Error PERMISSION_DENIED for user root on table BulkIT_test")); } From d1fb7fdaa67ade3172253adf482723cb2cbe2807 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Fri, 25 Sep 2026 13:48:20 +0000 Subject: [PATCH 3/6] Protect against NPE --- .../org/apache/accumulo/server/security/SecurityOperation.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 caba2b94286..9af9c0410d7 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 @@ -543,7 +543,8 @@ protected boolean canBulkImport(TCredentials c, TableId tableId, String tableNam final String allowedLocations = context.getTableConfiguration(tableId).get(Property.GENERAL_BULK_SOURCE_DIRS); boolean locationAllowed = false; - if (allowedLocations.isBlank() || Pattern.matches(allowedLocations, dir)) { + if (allowedLocations == null || allowedLocations.isBlank() + || Pattern.matches(allowedLocations, dir)) { locationAllowed = true; } if (!locationAllowed) { From 3d7a706724f7d77d24214b9e46d8f2ec439b4bd2 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Tue, 29 Sep 2026 07:34:25 -0400 Subject: [PATCH 4/6] Apply suggestion from @Amemeda Co-authored-by: Amanda Villarreal <48224036+Amemeda@users.noreply.github.com> --- core/src/main/java/org/apache/accumulo/core/conf/Property.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 690550faf13..e335ac1b206 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 @@ -315,7 +315,7 @@ public enum Property { "2.0.0"), GENERAL_BULK_SOURCE_DIRS("general.bulk.allowed.source.dir.pattern", "^(?!.*/accumulo/tables).*$", PropertyType.STRING, - "A regular expressions where source bulk import files for the table may located." + "A regular expression where source bulk import files for the table may be located." + " The default value prevents using rfiles from the default Accumulo table directory" + " path from being the source directory.", "2.1.7"), From 0ab808023e2c9c718f31f85ca396a8e728ed8834 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Tue, 20 Oct 2026 16:40:03 +0000 Subject: [PATCH 5/6] Implemented PR suggestions --- .../apache/accumulo/core/conf/Property.java | 9 +-- .../server/security/SecurityOperation.java | 28 ++++++- .../test/functional/BulkFailureIT.java | 1 + .../accumulo/test/functional/BulkIT.java | 77 +++++++++++++------ .../accumulo/test/functional/BulkNewIT.java | 2 + .../functional/BulkNewMetadataSkipIT.java | 5 +- .../accumulo/test/functional/BulkOldIT.java | 2 + .../functional/BulkSplitOptimizationIT.java | 1 + 8 files changed, 92 insertions(+), 33 deletions(-) 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 e335ac1b206..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,11 +313,9 @@ 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.allowed.source.dir.pattern", "^(?!.*/accumulo/tables).*$", - PropertyType.STRING, - "A regular expression where source bulk import files for the table may be located." - + " The default value prevents using rfiles from the default Accumulo table directory" - + " path from being the source directory.", + 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, @@ -2132,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 9af9c0410d7..059212abf78 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,8 +20,10 @@ 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; @@ -60,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 @@ -74,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; @@ -540,13 +548,25 @@ protected boolean canBulkImport(TCredentials c, TableId tableId, String tableNam authenticate(c); final boolean tablePerm = hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false); + final String allowedLocations = - context.getTableConfiguration(tableId).get(Property.GENERAL_BULK_SOURCE_DIRS); + context.getConfiguration().get(Property.GENERAL_BULK_SOURCE_DIRS); boolean locationAllowed = false; - if (allowedLocations == null || allowedLocations.isBlank() - || Pattern.matches(allowedLocations, dir)) { - locationAllowed = true; + try { + final String user = c.getPrincipal(); + 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 {} not allowed from directory {}, check table property {} value: {}", 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 67c5539b5f5..214d7a02e83 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 @@ -24,12 +24,15 @@ 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; @@ -39,6 +42,10 @@ 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 @@ -46,6 +53,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; @@ -54,19 +63,60 @@ protected Duration defaultTimeout() { return Duration.ofMinutes(4); } + 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")); } } @@ -110,25 +160,6 @@ static void runTest(AccumuloClient c, FileSystem fs, Path basePath, String table verifyParams.startRow = N; verifyParams.rows = 1; VerifyIngest.verifyIngest(c, verifyParams); - - // Create a subdirectory that is not allowed and try to bulk import from - // that location - Path a = new Path(base, "accumulo"); - fs.deleteOnExit(a); - Path t = new Path(a, "tables"); - fs.deleteOnExit(t); - fs.mkdirs(t); - params = new IngestParams(c.properties(), tableName, N); - params.outputFile = new Path(t, String.format(fileFormat, N)).toString(); - params.startRow = N; - params.rows = 1; - // create an rfile with one entry, there was a bug with this: - TestIngest.ingest(c, fs, params); - AccumuloSecurityException ase = assertThrows(AccumuloSecurityException.class, - () -> bulkLoad(c, tableName, bulkFailures, t, useOld)); - assertTrue( - ase.getMessage().contains("Error PERMISSION_DENIED for user root on table BulkIT_test")); - } @SuppressWarnings("deprecation") 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..1e14ab8b124 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, testDir); FunctionalTestUtils.createRFiles(c, fs, testDir.toString(), ROWS, SPLITS, 8); FileStatus[] stats = fs.listStatus(testDir); From e739a191b6a708291ecc6fda41bfc22fe2742615 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Wed, 21 Oct 2026 15:04:33 +0000 Subject: [PATCH 6/6] Fixed failing ITs --- .../server/security/SecurityOperation.java | 6 +++--- .../apache/accumulo/test/AuditMessageIT.java | 4 ++++ .../accumulo/test/BulkImportMonitoringIT.java | 3 +++ .../test/BulkImportSequentialRowsIT.java | 2 ++ .../accumulo/test/BulkImportVolumeIT.java | 2 ++ .../accumulo/test/CountNameNodeOpsBulkIT.java | 2 ++ .../accumulo/test/ScanConsistencyIT.java | 2 ++ .../ExternalCompactionProgressIT.java | 3 +++ .../accumulo/test/functional/BulkIT.java | 10 ++++++++++ .../functional/BulkSplitOptimizationIT.java | 2 +- .../test/functional/CompactionIT.java | 1 + .../accumulo/test/functional/SplitIT.java | 1 + .../accumulo/test/functional/SslIT.java | 1 + .../accumulo/test/shell/ShellServerIT.java | 19 +++++++++++++++++++ 14 files changed, 54 insertions(+), 4 deletions(-) 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 059212abf78..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 @@ -549,11 +549,11 @@ protected boolean canBulkImport(TCredentials c, TableId tableId, String tableNam 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 String user = c.getPrincipal(); final Map allowedLocationsMap = gson.fromJson(allowedLocations, jsonMapType); final String userSourceLocationPattern = allowedLocationsMap.get(user); if (userSourceLocationPattern == null) { @@ -569,8 +569,8 @@ protected boolean canBulkImport(TCredentials c, TableId tableId, String tableNam if (!locationAllowed) { log.error( - "Bulk Import into table {} not allowed from directory {}, check table property {} value: {}", - tableId, dir, Property.GENERAL_BULK_SOURCE_DIRS.getKey(), allowedLocations); + "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; } 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/BulkIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java index 214d7a02e83..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 @@ -22,6 +22,7 @@ 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; @@ -38,6 +39,7 @@ 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; @@ -63,6 +65,14 @@ 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<>(); 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 1e14ab8b124..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,7 +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, 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");