This is an automated email from the ASF dual-hosted git repository. dlmarion pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit a6d4e876b976c456623d3d90ab17c93a57fee7cd Merge: f6e82e6c89 130d0ed6a5 Author: Dave Marion <[email protected]> AuthorDate: Thu Jun 4 19:31:36 2026 +0000 Merge branch '2.1' .../accumulo/manager/tableOps/bulkVer2/BulkImportMove.java | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --cc server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/BulkImportMove.java index d86b270192,a1f258091e..4537cf3cd2 --- a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/BulkImportMove.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/BulkImportMove.java @@@ -26,13 -29,21 +26,16 @@@ import org.apache.accumulo.core.clientI import org.apache.accumulo.core.clientImpl.bulk.BulkSerialize; import org.apache.accumulo.core.clientImpl.thrift.TableOperation; import org.apache.accumulo.core.clientImpl.thrift.TableOperationExceptionType; -import org.apache.accumulo.core.conf.AccumuloConfiguration; -import org.apache.accumulo.core.conf.Property; -import org.apache.accumulo.core.fate.FateTxId; +import org.apache.accumulo.core.fate.FateId; import org.apache.accumulo.core.fate.Repo; import org.apache.accumulo.core.logging.BulkLogger; -import org.apache.accumulo.core.manager.state.tables.TableState; -import org.apache.accumulo.core.master.thrift.BulkImportState; -import org.apache.accumulo.manager.Manager; -import org.apache.accumulo.manager.tableOps.ManagerRepo; +import org.apache.accumulo.manager.tableOps.FateEnv; import org.apache.accumulo.server.fs.VolumeManager; -import org.apache.accumulo.server.zookeeper.TransactionWatcher.ZooArbitrator; +import org.apache.accumulo.server.util.bulkCommand.ListBulk.BulkState; + import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; + import org.slf4j.Logger; + import org.slf4j.LoggerFactory; /** * Bulk import makes requests of tablet servers, and those requests can take a long time. Our @@@ -53,8 -64,11 +56,10 @@@ class BulkImportMove extends AbstractBu private static final long serialVersionUID = 1L; + private static final Logger log = LoggerFactory.getLogger(BulkImportMove.class); - private final BulkInfo bulkInfo; + public BulkImportMove(BulkInfo bulkInfo) { - this.bulkInfo = bulkInfo; + super(bulkInfo); } @Override @@@ -62,12 -76,23 +67,18 @@@ final Path bulkDir = new Path(bulkInfo.bulkDir); final Path sourceDir = new Path(bulkInfo.sourceDir); - VolumeManager fs = manager.getVolumeManager(); + VolumeManager fs = env.getVolumeManager(); + if (log.isTraceEnabled()) { + FileStatus[] files = fs.listStatus(sourceDir); - log.trace("{} bulk import move starting. source:{} dest:{} files to move:{}", - FateTxId.formatTid(tid), sourceDir, bulkDir, files == null ? 0 : files.length); - } - - if (bulkInfo.tableState == TableState.ONLINE) { - ZooArbitrator.start(manager.getContext(), Constants.BULK_ARBITRATOR_TYPE, tid); ++ log.trace("{} bulk import move starting. source:{} dest:{} files to move:{}", fateId, ++ sourceDir, bulkDir, files == null ? 0 : files.length); + } + try { - manager.updateBulkImportStatus(sourceDir.toString(), BulkImportState.MOVING); Map<String,String> oldToNewNameMap = BulkSerialize.readRenameMap(bulkDir.toString(), fs::open); - moveFiles(tid, sourceDir, bulkDir, manager, fs, oldToNewNameMap); + moveFiles(fateId, sourceDir, bulkDir, env, fs, oldToNewNameMap); return new LoadFiles(bulkInfo); } catch (Exception ex) {
