Add an AsciiDoc generator for any ConfigDef class (#8)

* Fixup some setting description and error messages
* Add the AsciiDocGenerator utility class

Author: Vincent Royer

common/src/main/java/com/datastax/oss/common/sink/config/AsciiDocGenerator.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed 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
+ *
+ * http://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 com.datastax.oss.common.sink.config;
+
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.nio.file.Path;
+import java.util.*;
+import org.apache.kafka.common.config.ConfigDef;
+
+/** Generates Asiidoc documentation for a ConfigDef */
+public class AsciiDocGenerator {
+
+  public static String toAscidoc(ConfigDef configDef) {
+    StringBuilder b = new StringBuilder();
+    for (ConfigDef.ConfigKey key : sortedConfigs(configDef)) {
+      if (key.internalConfig) {
+        continue;
+      }
+      getConfigKeyAsciiDoc(configDef, key, b);
+      b.append("\n");
+    }
+    return b.toString();
+  }
+
+  protected static void getConfigKeyAsciiDoc(
+      ConfigDef configDef, ConfigDef.ConfigKey key, StringBuilder b) {
+    b.append("[#").append(key.name).append("]").append("\n");
+    for (String docLine : key.documentation.split("\n")) {
+      if (docLine.length() == 0) {
+        continue;
+      }
+      b.append(docLine).append("\n+\n");
+    }
+    b.append("Type: ").append(getConfigValue(key, "Type")).append("\n");
+    if (key.hasDefault()) {
+      b.append("Default: ").append(getConfigValue(key, "Default")).append("\n");
+    }
+    if (key.validator != null) {
+      b.append("Valid Values: ").append(getConfigValue(key, "Valid Values")).append("\n");
+    }
+    b.append("Importance: ").append(getConfigValue(key, "Importance")).append("\n");
+  }
+
+  protected static String getConfigValue(ConfigDef.ConfigKey key, String headerName) {
+    switch (headerName) {
+      case "Name":
+        return key.name;
+      case "Description":
+        return key.documentation;
+      case "Type":
+        return key.type.toString().toLowerCase(Locale.ROOT);
+      case "Default":
+        if (key.hasDefault()) {
+          if (key.defaultValue == null) return "null";
+          String defaultValueStr = ConfigDef.convertToString(key.defaultValue, key.type);
+          if (defaultValueStr.isEmpty()) return "\"\"";
+          else return defaultValueStr;
+        } else return "";
+      case "Valid Values":
+        return key.validator != null ? key.validator.toString() : "";
+      case "Importance":
+        return key.importance.toString().toLowerCase(Locale.ROOT);
+      default:
+        throw new RuntimeException(
+            "Can't find value for header '" + headerName + "' in " + key.name);
+    }
+  }
+
+  protected static List<ConfigDef.ConfigKey> sortedConfigs(ConfigDef configDef) {
+    final Map<String, Integer> groupOrd = new HashMap<>(configDef.groups().size());
+    int ord = 0;
+    for (String group : configDef.groups()) {
+      groupOrd.put(group, ord++);
+    }
+
+    List<ConfigDef.ConfigKey> configs = new ArrayList<>(configDef.configKeys().values());
+    Collections.sort(configs, (k1, k2) -> compare(k1, k2, groupOrd));
+    return configs;
+  }
+
+  protected static int compare(
+      ConfigDef.ConfigKey k1, ConfigDef.ConfigKey k2, Map<String, Integer> groupOrd) {
+    int cmp =
+        k1.group == null
+            ? (k2.group == null ? 0 : -1)
+            : (k2.group == null
+                ? 1
+                : Integer.compare(groupOrd.get(k1.group), groupOrd.get(k2.group)));
+    if (cmp == 0) {
+      cmp = Integer.compare(k1.orderInGroup, k2.orderInGroup);
+      if (cmp == 0) {
+        // first take anything with no default value
+        if (!k1.hasDefault() && k2.hasDefault()) cmp = -1;
+        else if (!k2.hasDefault() && k1.hasDefault()) cmp = 1;
+        else {
+          cmp = k1.importance.compareTo(k2.importance);
+          if (cmp == 0) return k1.name.compareTo(k2.name);
+        }
+      }
+    }
+    return cmp;
+  }
+
+  public static void generateConfigDefDoc(Path path, String name, String title, ConfigDef configDef)
+      throws IOException {
+    try (FileWriter fileWriter = new FileWriter(path.resolve(name).toFile())) {
+      PrintWriter pw = new PrintWriter(fileWriter);
+      pw.append("= ").append(title).append("\n\n");
+      pw.append("== Parameters").append("\n\n");
+      pw.append(toAscidoc(configDef));
+      pw.flush();
+    }
+  }
+}

common/src/main/java/com/datastax/oss/common/sink/config/AuthenticatorConfig.java

@@ -48,8 +48,9 @@ public class AuthenticatorConfig extends AbstractConfig {
               PROVIDER_OPT,
               ConfigDef.Type.STRING,
               "None",
+              ConfigDef.ValidString.in("None", "PLAIN", "GSSAPI"),
               ConfigDef.Importance.HIGH,
-              "None | PLAIN | GSSAPI")
+              "Authentication provider")
           .define(
               USERNAME_OPT,
               ConfigDef.Type.STRING,

common/src/main/java/com/datastax/oss/common/sink/config/CassandraSinkConfig.java

@@ -163,8 +163,9 @@ public class CassandraSinkConfig {
               COMPRESSION_OPT,
               ConfigDef.Type.STRING,
               "None",
+              ConfigDef.CaseInsensitiveValidString.in("NONE", "LZ4", "SNAPPY"),
               ConfigDef.Importance.HIGH,
-              "None | LZ4 | Snappy")
+              "Compression algorithm to use when issuing requests to the database server")
           .define(
               QUERY_EXECUTION_TIMEOUT_OPT,
               ConfigDef.Type.INT,

common/src/main/java/com/datastax/oss/common/sink/config/SslConfig.java

@@ -57,8 +57,9 @@ public class SslConfig extends AbstractConfig {
               PROVIDER_OPT,
               ConfigDef.Type.STRING,
               "None",
+              ConfigDef.ValidString.in("None", "JDK", "OpenSSL"),
               ConfigDef.Importance.HIGH,
-              "None | JDK | OpenSSL")
+              "SSL/TLS provider")
           .define(
               CIPHER_SUITES_OPT,
               ConfigDef.Type.LIST,

common/src/test/java/com/datastax/oss/common/sink/config/AuthenticatorConfigTest.java

@@ -151,10 +151,10 @@ void should_error_invalid_provider() {
     Map<String, String> props =
         ImmutableMap.<String, String>builder().put(PROVIDER_OPT, "foo").build();
     assertThatThrownBy(() -> new AuthenticatorConfig(props))
-        .isInstanceOf(ConfigException.class)
+        .isInstanceOf(org.apache.kafka.common.config.ConfigException.class)
         .hasMessage(
             String.format(
-                "Invalid value foo for configuration %s: valid values are None, PLAIN, GSSAPI",
+                "Invalid value foo for configuration %s: String must be one of: None, PLAIN, GSSAPI",
                 PROVIDER_OPT));
   }
 

common/src/test/java/com/datastax/oss/common/sink/config/CassandraSinkConfigTest.java

@@ -194,7 +194,7 @@ void should_error_invalid_compression_type() {
         .isInstanceOf(ConfigException.class)
         .hasMessageContaining(
             String.format(
-                "Invalid value foo for configuration %s: valid values are none, snappy, lz4",
+                "Invalid value foo for configuration %s: String must be one of (case insensitive): LZ4, NONE, SNAPPY",
                 COMPRESSION_OPT));
   }
 

common/src/test/java/com/datastax/oss/common/sink/config/SslConfigTest.java

@@ -211,9 +211,9 @@ void should_error_invalid_provider() {
     Map<String, String> props =
         ImmutableMap.<String, String>builder().put(PROVIDER_OPT, "foo").build();
     assertThatThrownBy(() -> new SslConfig(props))
-        .isInstanceOf(ConfigException.class)
+        .isInstanceOf(org.apache.kafka.common.config.ConfigException.class)
         .hasMessage(
             "Invalid value foo for configuration ssl.provider: "
-                + "valid values are None, JDK, OpenSSL");
+                + "String must be one of: None, JDK, OpenSSL");
   }
 }