Merge branch '3.1'

Author: keith-turner

core/src/main/java/org/apache/accumulo/core/conf/ConfigurationTypeHelper.java

@@ -18,18 +18,27 @@
  */
 package org.apache.accumulo.core.conf;
 
+import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.HOURS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
+import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -222,4 +231,46 @@ public static int getNumThreads(String threads) {
     }
     return nThreads;
   }
+
+  /**
+   * Get the set of volumes parsed from a volumes property type, and throw exceptions if the volumes
+   * aren't valid, are null, contains only blanks, or contains duplicates. An empty string is
+   * allowed (resulting in an empty set of volumes), to handle the case where the property is not
+   * set by a user (or... is set to the same as the default, which is equivalent to not being set).
+   * If the property is required to be set, it is the caller's responsibility to verify that the
+   * parsed set is non-empty.
+   *
+   * @throws IllegalArgumentException when the volumes are set to something that cannot be parsed
+   */
+  public static Set<String> getVolumeUris(String volumes) {
+    if (requireNonNull(volumes).isEmpty()) {
+      // special case when the property is not set and defaults to an empty string
+      return Set.of();
+    }
+    var blanksRemoved = Arrays.stream(volumes.split(",")).map(String::strip)
+        .filter(Predicate.not(String::isEmpty)).collect(Collectors.toList());
+    if (blanksRemoved.isEmpty()) {
+      throw new IllegalArgumentException("property contains only blank volumes");
+    }
+    var deduplicated = blanksRemoved.stream().map(ConfigurationTypeHelper::normalizeVolume)
+        .collect(Collectors.toCollection(LinkedHashSet::new));
+    if (deduplicated.size() < blanksRemoved.size()) {
+      throw new IllegalArgumentException("property contains duplicate volumes");
+    }
+    return deduplicated;
+  }
+
+  private static String normalizeVolume(String volume) {
+    if (!volume.contains(":")) {
+      throw new IllegalArgumentException("'" + volume + "' is not a fully qualified URI");
+    }
+    try {
+      // pass through URI to unescape hex encoded chars (e.g. convert %2C to "," char)
+      return new Path(new URI(volume.strip())).toString();
+    } catch (URISyntaxException e) {
+      throw new IllegalArgumentException(
+          "volume contains '" + volume + "' which has a syntax error", e);
+    }
+  }
+
 }

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -146,7 +146,7 @@ public enum Property {
           + " HDFS. To use the ChangeSecret tool, run the command: `./bin/accumulo"
           + " admin changeSecret`.",
       "1.3.5"),
-  INSTANCE_VOLUMES("instance.volumes", "", PropertyType.STRING,
+  INSTANCE_VOLUMES("instance.volumes", "", PropertyType.VOLUMES,
       "A comma separated list of dfs uris to use. Files will be stored across"
           + " these filesystems. In some situations, the first volume in this list"
           + " may be treated differently, such as being preferred for writing out"

core/src/main/java/org/apache/accumulo/core/conf/PropertyType.java

@@ -136,16 +136,19 @@ public enum PropertyType {
           + " interpreted based on the context of the property to which it applies."),
 
   JSON("json", new ValidJson(),
-      "An arbitrary string that is represents a valid, parsable generic json object."
-          + "The validity of the json object in the context of the property usage is not checked by this type."),
+      "An arbitrary string that is represents a valid, parsable generic json object. The validity "
+          + "of the json object in the context of the property usage is not checked by this type."),
+
   BOOLEAN("boolean", in(false, null, "true", "false"),
       "Has a value of either 'true' or 'false' (case-insensitive)"),
 
   URI("uri", x -> true, "A valid URI"),
 
   FILENAME_EXT("file name extension", in(true, RFile.EXTENSION),
       "One of the currently supported filename extensions for storing table data files. "
-          + "Currently, only " + RFile.EXTENSION + " is supported.");
+          + "Currently, only " + RFile.EXTENSION + " is supported."),
+
+  VOLUMES("volumes", new ValidVolumes(), "See instance.volumes documentation");
 
   private final String shortname;
   private final String format;
@@ -212,13 +215,32 @@ private static class ValidJson implements Predicate<String> {
     public boolean test(String value) {
       try {
         if (value.length() > ONE_MILLION) {
-          log.info("provided json string length {} is greater than limit of {} for parsing",
+          log.error("provided json string length {} is greater than limit of {} for parsing",
               value.length(), ONE_MILLION);
           return false;
         }
         jsonMapper.readTree(value);
         return true;
-      } catch (IOException ex) {
+      } catch (IOException e) {
+        log.error("provided json string resulted in an error", e);
+        return false;
+      }
+    }
+  }
+
+  private static class ValidVolumes implements Predicate<String> {
+    private static final Logger log = LoggerFactory.getLogger(ValidVolumes.class);
+
+    @Override
+    public boolean test(String volumes) {
+      if (volumes == null) {
+        return false;
+      }
+      try {
+        ConfigurationTypeHelper.getVolumeUris(volumes);
+        return true;
+      } catch (IllegalArgumentException e) {
+        log.error("provided volume string is not valid", e);
         return false;
       }
     }
@@ -392,5 +414,4 @@ public static IntStream parse(String portRange) {
     }
 
   }
-
 }

core/src/main/java/org/apache/accumulo/core/volume/VolumeConfiguration.java

@@ -19,14 +19,10 @@
 package org.apache.accumulo.core.volume;
 
 import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.Arrays;
-import java.util.LinkedHashSet;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -39,34 +35,12 @@ public static FileSystem fileSystemForPath(String path, Configuration conf) thro
   }
 
   public static Set<String> getVolumeUris(AccumuloConfiguration conf) {
-    String volumes = conf.get(Property.INSTANCE_VOLUMES);
+    var volumes = conf.get(Property.INSTANCE_VOLUMES);
     if (volumes == null || volumes.isBlank()) {
       throw new IllegalArgumentException(
           "Missing required property " + Property.INSTANCE_VOLUMES.getKey());
     }
-    String[] volArray = volumes.split(",");
-    LinkedHashSet<String> deduplicated =
-        Arrays.stream(volArray).map(VolumeConfiguration::normalizeVolume)
-            .collect(Collectors.toCollection(LinkedHashSet::new));
-    if (deduplicated.size() < volArray.length) {
-      throw new IllegalArgumentException(
-          Property.INSTANCE_VOLUMES.getKey() + " contains duplicate volumes (" + volumes + ")");
-    }
-    return deduplicated;
-  }
-
-  private static String normalizeVolume(String volume) {
-    if (volume == null || volume.isBlank() || !volume.contains(":")) {
-      throw new IllegalArgumentException("Expected fully qualified URI for "
-          + Property.INSTANCE_VOLUMES.getKey() + " got " + volume);
-    }
-    try {
-      // pass through URI to unescape hex encoded chars (e.g. convert %2C to "," char)
-      return new Path(new URI(volume.strip())).toString();
-    } catch (URISyntaxException e) {
-      throw new IllegalArgumentException(Property.INSTANCE_VOLUMES.getKey() + " contains '" + volume
-          + "' which has a syntax error", e);
-    }
+    return ConfigurationTypeHelper.getVolumeUris(volumes);
   }
 
 }

core/src/test/java/org/apache/accumulo/core/conf/ConfigurationTypeHelperTest.java

@@ -24,7 +24,9 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
@@ -132,4 +134,52 @@ public void testGetFractionFailureCase2() {
   public void testGetFractionFailureCase3() {
     assertThrows(IllegalArgumentException.class, () -> ConfigurationTypeHelper.getFraction(".%"));
   }
+
+  @Test
+  public void testGetVolumeUris() {
+    // test property not set
+    assertEquals(Set.of(), ConfigurationTypeHelper.getVolumeUris(""));
+
+    // test blank cases
+    assertThrows(NullPointerException.class, () -> ConfigurationTypeHelper.getVolumeUris(null));
+    for (String s : Set.of("   ", ",", ",,,", " ,,,", ",,, ", ", ,,")) {
+      var e = assertThrows(IllegalArgumentException.class,
+          () -> ConfigurationTypeHelper.getVolumeUris(s));
+      assertEquals("property contains only blank volumes", e.getMessage());
+    }
+
+    // test 1 volume
+    for (String s : Set.of("hdfs:/volA", ",hdfs:/volA", "hdfs:/volA,")) {
+      var uris = ConfigurationTypeHelper.getVolumeUris(s);
+      assertEquals(1, uris.size());
+      assertTrue(uris.contains("hdfs:/volA"));
+    }
+
+    // test more than 1 volume
+    for (String s : Set.of("hdfs:/volA,file:/volB", ",hdfs:/volA,file:/volB",
+        "hdfs:/volA,,file:/volB", "hdfs:/volA,file:/volB,   ,")) {
+      var uris = ConfigurationTypeHelper.getVolumeUris(s);
+      assertEquals(2, uris.size());
+      assertTrue(uris.contains("hdfs:/volA"));
+      assertTrue(uris.contains("file:/volB"));
+    }
+
+    // test invalid URI
+    for (String s : Set.of("hdfs:/volA,hdfs:/volB,volA", ",volA,hdfs:/volA,hdfs:/volB",
+        "hdfs:/volA,,volA,hdfs:/volB", "hdfs:/volA,volA,hdfs:/volB,   ,")) {
+      var iae = assertThrows(IllegalArgumentException.class,
+          () -> ConfigurationTypeHelper.getVolumeUris(s));
+      assertEquals("'volA' is not a fully qualified URI", iae.getMessage());
+    }
+
+    // test duplicates
+    var iae = assertThrows(IllegalArgumentException.class,
+        () -> ConfigurationTypeHelper.getVolumeUris("hdfs:/volA,hdfs:/volB,hdfs:/volA"));
+    assertEquals("property contains duplicate volumes", iae.getMessage());
+
+    // test syntax error in URI
+    iae = assertThrows(IllegalArgumentException.class,
+        () -> ConfigurationTypeHelper.getVolumeUris("hdfs:/volA,hdfs :/::/volB"));
+    assertEquals("volume contains 'hdfs :/::/volB' which has a syntax error", iae.getMessage());
+  }
 }

core/src/test/java/org/apache/accumulo/core/conf/PropertyTypeTest.java

@@ -219,4 +219,15 @@ public void testTypeFILENAME_EXT() {
     invalid(null, "RF", "map", "", "MAP", "rF", "Rf", " rf ");
   }
 
+  @Test
+  public void testTypeVOLUMES() {
+    // more comprehensive parsing tests are in ConfigurationTypeHelperTest.testGetVolumeUris()
+    valid("", "hdfs:/volA", ",hdfs:/volA", "hdfs:/volA,", "hdfs:/volA,file:/volB",
+        ",hdfs:/volA,file:/volB", "hdfs:/volA,,file:/volB", "hdfs:/volA,file:/volB,   ,");
+    invalid(null, "   ", ",", ",,,", " ,,,", ",,, ", ", ,,", "hdfs:/volA,hdfs:/volB,volA",
+        ",volA,hdfs:/volA,hdfs:/volB", "hdfs:/volA,,volA,hdfs:/volB",
+        "hdfs:/volA,volA,hdfs:/volB,   ,", "hdfs:/volA,hdfs:/volB,hdfs:/volA",
+        "hdfs:/volA,hdfs :/::/volB");
+  }
+
 }

core/src/test/java/org/apache/accumulo/core/volume/VolumeConfigurationTest.java

@@ -0,0 +1,42 @@
+/*
+ * 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.core.volume;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.accumulo.core.conf.ConfigurationCopy;
+import org.apache.accumulo.core.conf.Property;
+import org.junit.jupiter.api.Test;
+
+public class VolumeConfigurationTest {
+  @Test
+  public void testEmptyVolumes() {
+    ConfigurationCopy config = new ConfigurationCopy();
+    assertNull(config.get(Property.INSTANCE_VOLUMES.getKey()));
+    assertThrows(IllegalArgumentException.class, () -> VolumeConfiguration.getVolumeUris(config));
+
+    config.set(Property.INSTANCE_VOLUMES.getKey(), "");
+    assertThrows(IllegalArgumentException.class, () -> VolumeConfiguration.getVolumeUris(config));
+    config.set(Property.INSTANCE_VOLUMES.getKey(), " ");
+    assertThrows(IllegalArgumentException.class, () -> VolumeConfiguration.getVolumeUris(config));
+    config.set(Property.INSTANCE_VOLUMES.getKey(), ",");
+    assertThrows(IllegalArgumentException.class, () -> VolumeConfiguration.getVolumeUris(config));
+  }
+}