Adds validation for instance.volumes property (#5365)

* Move parsing code for volume properties to ConfigurationTypeHelper and
  add a unit test
* Make VolumeManager do an additional check to require the property to
  be set (could also be done elsewhere in other overall configuration
  checks)
* Added a validator to the PropertyType to verify the format when it is
  set
* Slightly modified some of the exception messages, so that it does not
  mention the instance.volumes property itself... since the PropertyType
  validation could apply to other properties of the same type


Co-authored-by: Christopher Tubbs <ctubbsii@apache.org>

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

@@ -135,7 +135,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

@@ -139,16 +139,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;
@@ -215,13 +218,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;
       }
     }
@@ -395,5 +417,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

@@ -225,4 +225,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));
+  }
+}