Fix for #2872. Makes exporttable command volume aware.

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/tableExport/WriteExportFiles.java

@@ -28,6 +28,8 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipOutputStream;
 
@@ -154,6 +156,8 @@ public void undo(long tid, Manager env) {
   public static void exportTable(VolumeManager fs, ServerContext context, String tableName,
       TableId tableID, String exportDir) throws Exception {
 
+    Set<String> volumeSet = new TreeSet<>();
+
     fs.mkdirs(new Path(exportDir));
     Path exportMetaFilePath = fs.getFileSystemByPath(new Path(exportDir))
         .makeQualified(new Path(exportDir, Constants.EXPORT_FILE));
@@ -186,22 +190,47 @@ public static void exportTable(VolumeManager fs, ServerContext context, String t
       dataOut.close();
       dataOut = null;
 
-      createDistcpFile(fs, exportDir, exportMetaFilePath, uniqueFiles);
+      // make a set of unique volumes from the map
+      for (String fileString : uniqueFiles.values()) {
+        String uniqueVolume = getVolumeFromString(fileString);
+        volumeSet.add(uniqueVolume);
+      }
 
+      // for each unique volume: get every matching entry in the map and send them to
+      // createDistcpFile method
+      for (String volumeString : volumeSet) {
+        Set<String> sortedVolumeSet = new TreeSet<>();
+        for (String rFileString : uniqueFiles.values()) {
+          String currentVolume = getVolumeFromString(rFileString);
+          if (currentVolume.equals(volumeString)) {
+            sortedVolumeSet.add(rFileString);
+          }
+        }
+        createDistcpFile(fs, exportDir, exportMetaFilePath, sortedVolumeSet, volumeString);
+      }
     } finally {
       if (dataOut != null) {
         dataOut.close();
       }
     }
   }
 
+  private static String getVolumeFromString(String searchString) {
+    String[] segmentArray = searchString.split("/");
+    return segmentArray[2];
+  }
+
   private static void createDistcpFile(VolumeManager fs, String exportDir, Path exportMetaFilePath,
-      Map<String,String> uniqueFiles) throws IOException {
-    BufferedWriter distcpOut = new BufferedWriter(
-        new OutputStreamWriter(fs.create(new Path(exportDir, "distcp.txt")), UTF_8));
+      Set<String> uniqueFiles, String volumeName) throws IOException {
+    if (volumeName.contains(":")) {
+      volumeName = volumeName.replace(":", "-");
+    }
+
+    BufferedWriter distcpOut = new BufferedWriter(new OutputStreamWriter(
+        fs.create(new Path(exportDir, "distcp-" + volumeName + ".txt")), UTF_8));
 
     try {
-      for (String file : uniqueFiles.values()) {
+      for (String file : uniqueFiles) {
         distcpOut.append(file);
         distcpOut.newLine();
       }

test/src/main/java/org/apache/accumulo/test/ExportTableCommandWitlMultipleVolumes.java

@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.accumulo.test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.Text;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExportTableCommandWitlMultipleVolumes extends AccumuloClusterHarness {
+
+  private static final Logger log = LoggerFactory.getLogger(ImportExportIT.class);
+
+  Path v1, v2;
+
+  public static String[] row_numbers = "1,2,3,4,5,6,7,8,9,10".split(",");
+
+  String baseDirStr = "";
+  String baseDir2Str = "";
+
+  @Override
+  protected Duration defaultTimeout() {
+    return Duration.ofMinutes(1);
+  }
+
+  @Override
+  public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+    File baseDir = cfg.getDir();
+    baseDirStr = baseDir.toString();
+
+    // get first volume name
+    String[] baseDirArray = baseDirStr.split("/");
+    String originalVolume = baseDirArray[2];
+
+    // get second volume name
+    String[] baseDir2Array = baseDirArray;
+    baseDir2Array[2] = baseDir2Array[2] + "2";
+    String secondVolume = baseDir2Array[2];
+
+    // make second volume base directory
+    for (String element : baseDir2Array) {
+      baseDir2Str = baseDir2Str + "/" + element;
+    }
+    File baseDir2 = new File(baseDir2Str);
+
+    File v1f = new File(baseDir, "volumes/v1");
+    File v2f = new File(baseDir2, "volumes/v2");
+
+    v1 = new Path("file://" + v1f.getAbsolutePath());
+    v2 = new Path("file://" + v2f.getAbsolutePath());
+
+    // Run MAC on two locations in the local file system
+    cfg.setProperty(Property.INSTANCE_VOLUMES, v1 + "," + v2);
+
+    // use raw local file system so walogs sync and flush will work
+    hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
+  }
+
+  @Test
+  public void testExportCommand() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+
+      final String tableName = getUniqueNames(1)[0];
+      client.tableOperations().create(tableName);
+
+      // add splits to table
+      SortedSet<Text> partitions = new TreeSet<>();
+      for (String s : row_numbers) {
+        partitions.add(new Text(s));
+      }
+      client.tableOperations().addSplits(tableName, partitions);
+
+      try (BatchWriter bw = client.createBatchWriter(tableName)) {
+        for (int i = 1; i <= 5000000; i++) {
+          Mutation m = new Mutation(Integer.toString(i));
+          if (i % 2 != 0) {
+            m.put(Integer.toString(i), "", String
+                .format("file://localhost:8020/accumulo/tables/1/default_tablet/I00000%dp.rf", i));
+          } else {
+            m.put(Integer.toString(i), "", String
+                .format("file://localhost:8020/accumulo/tables/2/default_tablet/I00000%dp.rf", i));
+          }
+          bw.addMutation(m);
+        }
+      }
+
+      client.tableOperations().compact(tableName, null, null, true, true);
+      client.tableOperations().flush(tableName, null, null, true);
+
+      Path outputDir = new Path(cluster.getTemporaryPath(), getClass().getName());
+      Path exportDir = new Path(outputDir, "export");
+      client.tableOperations().offline(tableName, true);
+      client.tableOperations().exportTable(tableName, exportDir.toString());
+
+      try (Scanner scanner = client.createScanner(MetadataTable.NAME, Authorizations.EMPTY)) {
+        scanner.setRange(new Range("1", "1<"));
+        scanner.fetchColumnFamily(MetadataSchema.TabletsSection.DataFileColumnFamily.NAME);
+
+        for (Map.Entry<Key,Value> entry : scanner) {
+          log.info("Key is: " + entry.getKey());
+          log.info("Value is: " + entry.getValue());
+          boolean inV1 = entry.getKey().getColumnQualifier().toString().contains(v1.toString());
+          boolean inV2 = entry.getKey().getColumnQualifier().toString().contains(v2.toString());
+          assertTrue(inV1 || inV2);
+        }
+      }
+
+      FileSystem fs = cluster.getFileSystem();
+      fs.deleteOnExit(v1);
+      fs.deleteOnExit(v2);
+    }
+  }
+}

test/src/main/java/org/apache/accumulo/test/shell/ShellServerIT.java

@@ -167,6 +167,9 @@ public void exporttableImporttable() throws Exception {
     final String table = getUniqueNames(1)[0];
     final String table2 = table + "2";
 
+    String[] pathTokens = rootPath.split("/");
+    String volumeName = pathTokens[2];
+
     // exporttable / importtable
     ts.exec("createtable " + table + " -evc", true);
     make10();
@@ -195,8 +198,8 @@ public void exporttableImporttable() throws Exception {
       fs.mkdirs(importDir);
 
       // Implement a poor-man's DistCp
-      try (BufferedReader reader =
-          new BufferedReader(new FileReader(new File(exportDir, "distcp.txt"), UTF_8))) {
+      try (BufferedReader reader = new BufferedReader(
+          new FileReader(new File(exportDir, "distcp-" + volumeName + ".txt"), UTF_8))) {
         for (String line; (line = reader.readLine()) != null;) {
           Path exportedFile = new Path(line);
           // There isn't a cp on FileSystem??
@@ -208,7 +211,7 @@ public void exporttableImporttable() throws Exception {
         }
       }
     } else {
-      String[] distCpArgs = {"-f", exportUri + "/distcp.txt", import_};
+      String[] distCpArgs = {"-f", exportUri + "/distcp-" + volumeName + ".txt", import_};
       assertEquals(0, cp.run(distCpArgs), "Failed to run distcp: " + Arrays.toString(distCpArgs));
     }
     ts.exec("importtable " + table2 + " " + import_, true);
@@ -221,6 +224,66 @@ public void exporttableImporttable() throws Exception {
     ts.exec("deletetable -f " + table2, true);
   }
 
+  @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "path provided by test")
+  @Test
+  public void exporttableWithMultipleVolumes() throws Exception {
+
+    try (AccumuloClient client =
+        getCluster().createAccumuloClient(getPrincipal(), new PasswordToken(getRootPassword()))) {
+      client.securityOperations().grantNamespacePermission(getPrincipal(), "",
+          NamespacePermission.ALTER_NAMESPACE);
+    }
+
+    String[] pathTokens = rootPath.split("/");
+    String volumeName = pathTokens[2];
+
+    // exporttable
+    ts.exec("createtable multVolTable -evc", true);
+    makeTableWithMultipleVolumes();
+    // read table and get volumes!!
+    ts.exec("offline multVolTable", true);
+    File exportDir = new File(rootPath, "ShellServerIT.export");
+    String exportUri = "file://" + exportDir;
+    String localTmp = "file://" + new File(rootPath, "ShellServerIT.tmp");
+    ts.exec("exporttable -t multVolTable" + " " + exportUri, true);
+    DistCp cp = new DistCp(new Configuration(false), null);
+    String import_ = "file://" + new File(rootPath, "ShellServerIT.import");
+    ClientInfo info = ClientInfo.from(getCluster().getClientProperties());
+    if (info.saslEnabled()) {
+      // DistCp bugs out trying to get a fs delegation token to perform the cp. Just copy it
+      // ourselves by hand.
+      FileSystem fs = getCluster().getFileSystem();
+      FileSystem localFs = FileSystem.getLocal(new Configuration(false));
+
+      // Path on local fs to cp into
+      Path localTmpPath = new Path(localTmp);
+      localFs.mkdirs(localTmpPath);
+
+      // Path in remote fs to importtable from
+      Path importDir = new Path(import_);
+      fs.mkdirs(importDir);
+
+      // Implement a poor-man's DistCp
+      try (BufferedReader reader = new BufferedReader(
+          new FileReader(new File(exportDir, "distcp-" + volumeName + ".txt"), UTF_8))) {
+        for (String line; (line = reader.readLine()) != null;) {
+          Path exportedFile = new Path(line);
+          // There isn't a cp on FileSystem??
+          log.info("Copying {} to {}", line, localTmpPath);
+          fs.copyToLocalFile(exportedFile, localTmpPath);
+          Path tmpFile = new Path(localTmpPath, exportedFile.getName());
+          log.info("Moving {} to the import directory {}", tmpFile, importDir);
+          fs.moveFromLocalFile(tmpFile, importDir);
+        }
+      }
+    } else {
+      String[] distCpArgs = {"-f", exportUri + "/distcp-" + volumeName + ".txt", import_};
+      assertEquals(0, cp.run(distCpArgs), "Failed to run distcp: " + Arrays.toString(distCpArgs));
+    }
+    ts.exec("online multVolTable", true);
+    // ts.exec("deletetable -f multVolTable", true);
+  }
+
   @Test
   public void setscaniterDeletescaniter() throws Exception {
     final String table = getUniqueNames(1)[0];
@@ -1968,6 +2031,29 @@ private void make10() throws IOException {
     }
   }
 
+  private void makeTableWithMultipleVolumes() throws IOException {
+    for (int i = 1; i <= 4; i++) {
+      ts.exec(String.format(
+          "insert row%d cf I00000%dp.rf hdfs://warehouse-a.com:6093/accumulo/tables/3/default_tablet/I00000%dp.rf",
+          i, i, i));
+    }
+    for (int j = 5; j <= 8; j++) {
+      ts.exec(String.format(
+          "insert row%d cf J00000%dp.rf hdfs://n1.example.com:6093/accumulo/tables/3/default_tablet/J00000%dp.rf",
+          j, j, j));
+    }
+    for (int k = 9; k <= 12; k++) {
+      ts.exec(String.format(
+          "insert row%d cf K00000%dp.rf hdfs://warehouse-b.com:6093/accumulo/tables/3/default_tablet/K00000%dp.rf",
+          k, k, k));
+    }
+    for (int l = 13; l <= 16; l++) {
+      ts.exec(String.format(
+          "insert row%d cf L00000%dp.rf hdfs://n1.example.com:6090/accumulo/tables/3/default_tablet/L00000%dp.rf",
+          l, l, l));
+    }
+  }
+
   private List<String> getFiles(String tableId) throws IOException {
     ts.output.clear();