Update ServerConfiguration to use ConfigKey and ConfigKeyGroup

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/conf/ConfigKey.java

@@ -25,6 +25,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Function;
 import lombok.Builder;
 import lombok.Builder.Default;
 import lombok.Data;
@@ -100,7 +101,32 @@ public static ConfigKeyBuilder builder(String name) {
     @Default
     private Object defaultValue = null;
 
-    private String defaultValueAsString() {
+    /**
+     * Default value supplier.
+     */
+    @Default
+    private Function<Configuration, Object> defaultValueSupplier = (conf) -> null;
+
+    private Object defaultValue(Configuration conf) {
+        Object value = defaultValue();
+        if (null == value) {
+            return defaultValueSupplier.apply(conf);
+        } else {
+            return value;
+        }
+    }
+
+    private Number defaultValueAsNumber(Configuration conf) {
+        Object value = defaultValue(conf);
+        if (value instanceof Number) {
+            return (Number) value;
+        } else {
+            throw new IllegalArgumentException("The default value of key '" + name() + "' is not a NUMBER");
+        }
+    }
+
+    private String defaultValueAsString(Configuration conf) {
+        Object defaultValue = defaultValue(conf);
         if (null == defaultValue) {
             return null;
         } else if (defaultValue instanceof String) {
@@ -211,20 +237,25 @@ public void validate(Configuration conf) throws ConfigException {
      * @param value value of the setting
      */
     public void set(Configuration conf, Object value) {
-        if (!type().validator().validate(name(), value)) {
-            throw new IllegalArgumentException(
-                "Invalid value '" + value + "' to set on setting '" + name() + "': expected type = " + type);
-        }
+        if (null == value) {
+            conf.setProperty(name(), null);
+        } else {
+            if (!type().validator().validate(name(), value)) {
+                throw new IllegalArgumentException(
+                        "Invalid value '" + value + "' to set on setting '" + name() + "': expected type = " + type);
+            }
 
-        if (null != validator() && !validator().validate(name(), value)) {
-            throw new IllegalArgumentException(
-                "Invalid value '" + value + "' to set on setting '" + name() + "': required '" + validator() + "'");
-        }
+            if (null != validator() && !validator().validate(name(), value)) {
+                throw new IllegalArgumentException(
+                        "Invalid value '" + value + "' to set on setting '" + name()
+                                + "': required '" + validator() + "'");
+            }
 
-        if (value instanceof Class) {
-            conf.setProperty(name(), ((Class) value).getName());
-        } else {
-            conf.setProperty(name(), value);
+            if (value instanceof Class) {
+                conf.setProperty(name(), ((Class) value).getName());
+            } else {
+                conf.setProperty(name(), value);
+            }
         }
     }
 
@@ -236,7 +267,7 @@ public void set(Configuration conf, Object value) {
      */
     public long getLong(Configuration conf) {
         checkArgument(type() == Type.LONG, "'" + name() + "' is NOT a LONG numeric setting");
-        return conf.getLong(name(), (Long) defaultValue());
+        return conf.getLong(name(), defaultValueAsNumber(conf).longValue());
     }
 
     /**
@@ -247,7 +278,7 @@ public long getLong(Configuration conf) {
      */
     public int getInt(Configuration conf) {
         checkArgument(type() == Type.INT, "'" + name() + "' is NOT a INT numeric setting");
-        return conf.getInt(name(), (Integer) defaultValue());
+        return conf.getInt(name(), defaultValueAsNumber(conf).intValue());
     }
 
     /**
@@ -258,7 +289,7 @@ public int getInt(Configuration conf) {
      */
     public short getShort(Configuration conf) {
         checkArgument(type() == Type.SHORT, "'" + name() + "' is NOT a SHORT numeric setting");
-        return conf.getShort(name(), (Short) defaultValue());
+        return conf.getShort(name(), defaultValueAsNumber(conf).shortValue());
     }
 
     /**
@@ -269,7 +300,7 @@ public short getShort(Configuration conf) {
      */
     public boolean getBoolean(Configuration conf) {
         checkArgument(type() == Type.BOOLEAN, "'" + name() + "' is NOT a BOOL numeric setting");
-        return conf.getBoolean(name(), (Boolean) defaultValue());
+        return conf.getBoolean(name(), (Boolean) defaultValue(conf));
     }
 
     /**
@@ -280,7 +311,18 @@ public boolean getBoolean(Configuration conf) {
      */
     public double getDouble(Configuration conf) {
         checkArgument(type() == Type.DOUBLE, "'" + name() + "' is NOT a DOUBLE numeric setting");
-        return conf.getDouble(name(), (Double) defaultValue());
+        return conf.getDouble(name(), defaultValueAsNumber(conf).doubleValue());
+    }
+
+    /**
+     * Retrieve the setting from the configuration <tt>conf</tt> as a {@link Float} value.
+     *
+     * @param conf configuration to retrieve the setting
+     * @return the value as a float number
+     */
+    public float getFloat(Configuration conf) {
+        checkArgument(type() == Type.FLOAT, "'" + name() + "' is NOT a FLOAT numeric setting");
+        return conf.getFloat(name(), defaultValueAsNumber(conf).floatValue());
     }
 
     /**
@@ -290,7 +332,18 @@ public double getDouble(Configuration conf) {
      * @return the value as a string.
      */
     public String getString(Configuration conf) {
-        return conf.getString(name(), defaultValueAsString());
+        return conf.getString(name(), defaultValueAsString(conf));
+    }
+
+    /**
+     * Retrieve the setting from the configuration <tt>conf</tt> as a {@link String} value,
+     * if the setting isn't set, return <tt>null</tt>.
+     *
+     * @param conf configuration to retrieve the setting
+     * @return the value as a string
+     */
+    public String getStringWithoutDefault(Configuration conf) {
+        return conf.getString(name());
     }
 
     /**
@@ -303,7 +356,7 @@ public String getString(Configuration conf) {
     public <T> Class<? extends T> getClass(Configuration conf, Class<T> interfaceCls) {
         checkArgument(type() == Type.CLASS, "'" + name() + "' is NOT a CLASS setting");
         try {
-            Class<? extends T> defaultClass = (Class<? extends T>) defaultValue();
+            Class<? extends T> defaultClass = (Class<? extends T>) defaultValue(conf);
             return ReflectionUtils.getClass(conf, name(), defaultClass, interfaceCls, getClass().getClassLoader());
         } catch (ConfigurationException e) {
             throw new IllegalArgumentException("Invalid class is set to setting '" + name() + "': ", e);
@@ -320,7 +373,7 @@ public <T> Class<? extends T> getClass(Configuration conf, Class<T> interfaceCls
     public Class<?> getClass(Configuration conf) {
         checkArgument(type() == Type.CLASS, "'" + name() + "' is NOT a CLASS setting");
         try {
-            Class<?> defaultClass = (Class<?>) defaultValue();
+            Class<?> defaultClass = (Class<?>) defaultValue(conf);
             return ReflectionUtils.getClass(conf, name(), defaultClass, getClass().getClassLoader());
         } catch (ConfigurationException e) {
             throw new IllegalArgumentException("Invalid class is set to setting '" + name() + "': ", e);
@@ -336,13 +389,51 @@ public Class<?> getClass(Configuration conf) {
     @SuppressWarnings("unchecked")
     public List<Object> getList(Configuration conf) {
         checkArgument(type() == Type.LIST, "'" + name() + "' is NOT a LIST setting");
-        List<Object> list = (List<Object>) defaultValue();
+        List<Object> list = (List<Object>) defaultValue(conf);
         if (null == list) {
             list = Collections.emptyList();
         }
         return conf.getList(name(), list);
     }
 
+    /**
+     * Retrieve the setting from the configuration <tt>conf</tt> as a string array.
+     *
+     * @param conf configuration to retrieve the setting
+     * @return the value as a string array
+     * @see #getArrayWithoutDefault(Configuration)
+     */
+    public String[] getArray(Configuration conf) {
+        String[] retArray = getArrayWithoutDefault(conf);
+        if (null == retArray || retArray.length == 0) {
+            Object dv = defaultValue(conf);
+            if (dv instanceof String[]) {
+                return (String[]) dv;
+            } else {
+                return new String[] { defaultValueAsString(conf) };
+            }
+        } else {
+            return retArray;
+        }
+    }
+
+    /**
+     * Retrieve the setting from the configuration <tt>conf</tt> as a string array.
+     * If the key doesn't exist, it doesn't return the default value.
+     *
+     * @param conf configuration to retrieve the setting
+     * @return the value as a string array
+     * @see #getArray(Configuration)
+     */
+    public String[] getArrayWithoutDefault(Configuration conf) {
+        checkArgument(type() == Type.ARRAY, "'" + name() + "' is NOT an ARRAY setting");
+        if (conf.containsKey(name())) {
+            return conf.getStringArray(name());
+        } else {
+            return null;
+        }
+    }
+
     /**
      * Retrieve the setting value from the provided <tt>conf</tt>.
      *
@@ -358,10 +449,14 @@ public Object get(Configuration conf) {
                 return getShort(conf);
             case DOUBLE:
                 return getDouble(conf);
+            case FLOAT:
+                return getFloat(conf);
             case BOOLEAN:
                 return getBoolean(conf);
             case LIST:
                 return getList(conf);
+            case ARRAY:
+                return getArray(conf);
             case CLASS:
                 return getClass(conf);
             default:

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/conf/Type.java

@@ -33,8 +33,10 @@ public enum Type {
     INT((name, value) -> value instanceof Integer),
     SHORT((name, value) -> value instanceof Short),
     LONG((name, value) -> value instanceof Long),
+    FLOAT((name, value) -> value instanceof Float),
     DOUBLE((name, value) -> value instanceof Double),
-    LIST((name, value) -> value instanceof List),
+    LIST((name, value) -> value instanceof List || value instanceof String),
+    ARRAY((name, value) -> value instanceof String[] || value instanceof String),
     CLASS((name, value) -> value instanceof Class || value instanceof String);
 
     private Validator validator;

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/conf/Validator.java

@@ -45,4 +45,19 @@ default String documentation() {
         return "";
     }
 
+    default Validator and(Validator validator) {
+        return (name, value) ->
+            validate(name, value) && validator.validate(name, value);
+    }
+
+    default Validator or(Validator validator) {
+        return (name, value) ->
+            validate(name, value) || validator.validate(name, value);
+    }
+
+    default Validator negate() {
+        return (name, value) ->
+            !validate(name, value);
+    }
+
 }

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/conf/ConfigKeyTest.java

@@ -19,6 +19,7 @@
 
 package org.apache.bookkeeper.common.conf;
 
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -223,6 +224,29 @@ public void testGetDouble() {
         assertEquals(newValue, key.get(conf));
     }
 
+    @Test
+    public void testGetFloat() {
+        String keyName = runtime.getMethodName();
+        float defaultValue = ThreadLocalRandom.current().nextFloat();
+        ConfigKey key = ConfigKey.builder(keyName)
+            .required(true)
+            .type(Type.FLOAT)
+            .defaultValue(defaultValue)
+            .build();
+
+        Configuration conf = new ConcurrentConfiguration();
+
+        // get default value
+        assertEquals(defaultValue, key.getFloat(conf), 0.0001);
+        assertEquals(defaultValue, key.get(conf));
+
+        // set value
+        float newValue = (defaultValue * 2);
+        key.set(conf, newValue);
+        assertEquals(newValue, key.getFloat(conf), 0.0001);
+        assertEquals(newValue, key.get(conf));
+    }
+
     @Test
     public void testGetBoolean() {
         String keyName = runtime.getMethodName();
@@ -281,6 +305,41 @@ public void testGetList() {
         assertEquals(newList, key.get(conf));
     }
 
+    @Test
+    public void testGetArray() {
+        String keyName = runtime.getMethodName();
+        String[] defaultArray = new String[] {
+            "item1", "item2", "item3"
+        };
+        ConfigKey key = ConfigKey.builder(keyName)
+            .required(true)
+            .type(Type.ARRAY)
+            .defaultValue(defaultArray)
+            .build();
+
+        Configuration conf = new CompositeConfiguration();
+
+        // get default value
+        assertArrayEquals(defaultArray, key.getArray(conf));
+        assertArrayEquals(defaultArray, (String[]) key.get(conf));
+
+        // set value
+        String[] newArray = new String[] {
+            "item4", "item5", "item6"
+        };
+        key.set(conf, newArray);
+        assertArrayEquals(newArray, key.getArray(conf));
+        assertArrayEquals(newArray, (String[]) key.get(conf));
+
+        // set string value
+        newArray = new String[] {
+            "item7", "item8", "item9"
+        };
+        conf.setProperty(key.name(), "item7,item8,item9");
+        assertArrayEquals(newArray, key.getArray(conf));
+        assertArrayEquals(newArray, (String[]) key.get(conf));
+    }
+
     @Test
     public void testGetClass() {
         String keyName = runtime.getMethodName();

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieShell.java

@@ -2211,12 +2211,12 @@ private int updateBookieIdInCookie(final String bookieId, final boolean useHostn
                         for (File dir : ledgerDirectories) {
                             newCookie.writeToDirectory(dir);
                         }
-                        LOG.info("Updated cookie file present in ledgerDirectories {}", ledgerDirectories);
+                        LOG.info("Updated cookie file present in ledgerDirectories {}", (Object[]) ledgerDirectories);
                         if (ledgerDirectories != indexDirectories) {
                             for (File dir : indexDirectories) {
                                 newCookie.writeToDirectory(dir);
                             }
-                            LOG.info("Updated cookie file present in indexDirectories {}", indexDirectories);
+                            LOG.info("Updated cookie file present in indexDirectories {}", (Object[]) indexDirectories);
                         }
                     }
                     // writes newcookie to zookeeper