Fixed error caused during merge of #5355 to main (#5360)

When merging #5355 to main the FILES column ended up
not being fetched for the tablet metadata because the
Loader.pauseLimit variable had not been set. This variable
was set in the call to Loader.start, which happens after
the TabletsMetadata is configured

Author: dlmarion

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFiles.java

@@ -34,6 +34,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
@@ -115,7 +116,7 @@ public long isReady(FateId fateId, Manager manager) throws Exception {
     try (LoadMappingIterator lmi =
         BulkSerialize.getUpdatedLoadMapping(bulkDir.toString(), bulkInfo.tableId, fs::open)) {
 
-      Loader loader = new Loader();
+      Loader loader = new Loader(manager, bulkInfo.tableId);
 
       List<ColumnType> fetchCols = new ArrayList<>(List.of(PREV_ROW, LOCATION, LOADED, TIME));
       if (loader.pauseLimit > 0) {
@@ -137,27 +138,32 @@ public Repo<Manager> call(final FateId fateId, final Manager manager) {
 
   // visible for testing
   public static class Loader {
-    protected Path bulkDir;
-    protected Manager manager;
-    protected FateId fateId;
-    protected boolean setTime;
+    private final Manager manager;
+    private final long pauseLimit;
+
+    private Path bulkDir;
+    private FateId fateId;
+    private boolean setTime;
     Ample.ConditionalTabletsMutator conditionalMutator;
     private Map<KeyExtent,List<TabletFile>> loadingFiles;
-
     private long skipped = 0;
-    private long pauseLimit;
+
+    public Loader(Manager manager, TableId tableId) {
+      Objects.requireNonNull(manager, "Manager must be supplied");
+      Objects.requireNonNull(tableId, "Table ID must be supplied");
+      this.manager = manager;
+      this.pauseLimit =
+          manager.getContext().getTableConfiguration(tableId).getCount(Property.TABLE_FILE_PAUSE);
+    }
 
     void start(Path bulkDir, Manager manager, TableId tableId, FateId fateId, boolean setTime)
         throws Exception {
       this.bulkDir = bulkDir;
-      this.manager = manager;
       this.fateId = fateId;
       this.setTime = setTime;
       conditionalMutator = manager.getContext().getAmple().conditionallyMutateTablets();
       this.skipped = 0;
       this.loadingFiles = new HashMap<>();
-      this.pauseLimit =
-          manager.getContext().getTableConfiguration(tableId).getCount(Property.TABLE_FILE_PAUSE);
     }
 
     void load(List<TabletMetadata> tablets, Files files) {

server/manager/src/test/java/org/apache/accumulo/manager/tableOps/bulkVer2/LoadFilesTest.java

@@ -33,6 +33,7 @@
 import org.apache.accumulo.core.clientImpl.bulk.Bulk.FileInfo;
 import org.apache.accumulo.core.clientImpl.bulk.Bulk.Files;
 import org.apache.accumulo.core.clientImpl.bulk.LoadMappingIterator;
+import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.fate.FateId;
@@ -42,6 +43,8 @@
 import org.apache.accumulo.manager.Manager;
 import org.apache.accumulo.manager.tableOps.bulkVer2.LoadFiles.ImportTimingStats;
 import org.apache.accumulo.manager.tableOps.bulkVer2.LoadFiles.TabletsMetadataFactory;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.conf.TableConfiguration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
 import org.easymock.EasyMock;
@@ -60,6 +63,10 @@ public TestTabletsMetadata(AutoCloseable closeable, Iterable<TabletMetadata> tmi
 
   private static class CaptureLoader extends LoadFiles.Loader {
 
+    public CaptureLoader(Manager manager, TableId tableId) {
+      super(manager, tableId);
+    }
+
     private static class LoadResult {
       private final List<TabletMetadata> tablets;
       private final Files files;
@@ -159,19 +166,27 @@ public void testFindOverlappingFiles() {
   private Map<String,HashSet<KeyExtent>> runLoadFilesLoad(Map<KeyExtent,String> loadRanges)
       throws Exception {
 
+    Manager manager = EasyMock.createMock(Manager.class);
+    ServerContext ctx = EasyMock.createMock(ServerContext.class);
+    TableConfiguration tconf = EasyMock.createMock(TableConfiguration.class);
+
+    EasyMock.expect(manager.getContext()).andReturn(ctx).anyTimes();
+    EasyMock.expect(ctx.getTableConfiguration(tid)).andReturn(tconf).anyTimes();
+    EasyMock.expect(tconf.getCount(Property.TABLE_FILE_PAUSE))
+        .andReturn(Integer.parseInt(Property.TABLE_FILE_PAUSE.getDefaultValue())).anyTimes();
+
+    Path bulkDir = EasyMock.createMock(Path.class);
+    EasyMock.replay(manager, ctx, tconf, bulkDir);
+
     TabletsMetadata tabletMeta = new TestTabletsMetadata(null, tm);
     LoadMappingIterator lmi = PrepBulkImportTest.createLoadMappingIter(loadRanges);
-    CaptureLoader cl = new CaptureLoader();
+    CaptureLoader cl = new CaptureLoader(manager, tid);
     BulkInfo info = new BulkInfo();
     TabletsMetadataFactory tmf = (startRow) -> tabletMeta;
     FateId txid = FateId.from(FateInstanceType.USER, UUID.randomUUID());
 
-    Manager manager = EasyMock.createMock(Manager.class);
-    Path bulkDir = EasyMock.createMock(Path.class);
-    EasyMock.replay(manager, bulkDir);
-
     LoadFiles.loadFiles(cl, info, bulkDir, lmi, tmf, manager, txid);
-    EasyMock.verify(manager, bulkDir);
+    EasyMock.verify(manager, ctx, tconf, bulkDir);
     List<CaptureLoader.LoadResult> results = cl.getLoadResults();
     assertEquals(loadRanges.size(), results.size());