[#27] feat(core): add config system (#29)

### What changes were proposed in this pull request?

This PR introduces a Spark-style like strong typed config system for the
project.

### Why are the changes needed?

The config system is a cornerstone of the project. By comparing
different config system implementations, Spark's one is strong type
verified, and also can extend to other functions, so introducing a
Spark-style like config system for the project.

Fix: #27 

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?

Add new UTs to cover the test

core/src/main/java/com/datastrato/graviton/Config.java

@@ -0,0 +1,116 @@
+package com.datastrato.graviton;
+
+import com.datastrato.graviton.config.ConfigEntry;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class Config {
+
+  private static final Logger LOG = LoggerFactory.getLogger(Config.class);
+
+  private static final String CONFIG_PREPEND = "graviton.";
+
+  private final ConcurrentMap<String, String> configMap;
+
+  public Config(boolean loadDefaults) {
+    configMap = new ConcurrentHashMap<>();
+    if (loadDefaults) {
+      loadFromProperties(System.getProperties());
+    }
+  }
+
+  public Config() {
+    this(true);
+  }
+
+  public Config loadFromFile(String name) throws IOException {
+    String confDir =
+        Optional.ofNullable(System.getenv("GRAVITON_CONF_DIR"))
+            .orElse(
+                Optional.ofNullable(System.getenv("GRAVITON_HOME"))
+                    .map(s -> s + File.separator + "conf")
+                    .orElse(null));
+
+    if (confDir == null) {
+      throw new IllegalArgumentException("GRAVITON_CONF_DIR or GRAVITON_HOME not set");
+    }
+
+    File confFile = new File(confDir + File.separator + name);
+    if (!confFile.exists()) {
+      throw new IllegalArgumentException(
+          "Config file " + confFile.getAbsolutePath() + " not found");
+    }
+
+    Properties properties = loadPropertiesFromFile(confFile);
+    loadFromProperties(properties);
+
+    return this;
+  }
+
+  public <T> T get(ConfigEntry<T> entry) throws NoSuchElementException {
+    if (entry.isDeprecated()) {
+      LOG.warn("Config {} is deprecated.", entry.getKey());
+      if (!entry.getAlternatives().isEmpty()) {
+        LOG.warn("Please use {} instead.", String.join(", ", entry.getAlternatives()));
+      }
+    }
+
+    return entry.readFrom(configMap);
+  }
+
+  public <T> void set(ConfigEntry<T> entry, T value) {
+    if (entry.isDeprecated()) {
+      LOG.warn("Config {} is deprecated.", entry.getKey());
+      if (!entry.getAlternatives().isEmpty()) {
+        LOG.warn("Please use {} instead.", String.join(", ", entry.getAlternatives()));
+      }
+    }
+
+    if (value == null) {
+      LOG.warn("Config {} value to set is null, ignore setting to Config.", entry.getKey());
+    }
+
+    entry.writeTo(configMap, value);
+  }
+
+  private void loadFromMap(Map<String, String> map) {
+    map.forEach(
+        (k, v) -> {
+          String trimmedK = k.trim();
+          String trimmedV = v.trim();
+          if (!trimmedK.isEmpty() && !trimmedV.isEmpty() && trimmedK.startsWith(CONFIG_PREPEND)) {
+            configMap.put(trimmedK, trimmedV);
+          }
+        });
+  }
+
+  @VisibleForTesting
+  void loadFromProperties(Properties properties) {
+    loadFromMap(Maps.fromProperties(properties));
+  }
+
+  @VisibleForTesting
+  Properties loadPropertiesFromFile(File file) throws IOException {
+    Properties properties = new Properties();
+    try (InputStream in = Files.newInputStream(file.toPath())) {
+      properties.load(in);
+      return properties;
+
+    } catch (Exception e) {
+      LOG.error("Failed to load properties from " + file.getAbsolutePath(), e);
+      throw new IOException("Failed to load properties from " + file.getAbsolutePath(), e);
+    }
+  }
+}

core/src/main/java/com/datastrato/graviton/config/ConfigBuilder.java

@@ -0,0 +1,125 @@
+package com.datastrato.graviton.config;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+
+public class ConfigBuilder {
+
+  private String key;
+
+  private List<String> alternatives;
+
+  private String doc;
+
+  private String version;
+
+  private boolean isPublic;
+
+  private boolean isDeprecated;
+
+  public ConfigBuilder(String key) {
+    this.key = key;
+
+    this.alternatives = Collections.emptyList();
+    this.doc = "";
+    this.version = "0.1.0";
+    this.isPublic = true;
+    this.isDeprecated = false;
+  }
+
+  public ConfigBuilder alternatives(List<String> alternatives) {
+    this.alternatives = alternatives;
+    return this;
+  }
+
+  public ConfigBuilder doc(String doc) {
+    this.doc = doc;
+    return this;
+  }
+
+  public ConfigBuilder version(String version) {
+    this.version = version;
+    return this;
+  }
+
+  public ConfigBuilder internal() {
+    this.isPublic = false;
+    return this;
+  }
+
+  public ConfigBuilder deprecated() {
+    this.isDeprecated = true;
+    return this;
+  }
+
+  public ConfigEntry<String> stringConf() {
+    ConfigEntry<String> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    conf.setValueConverter(s -> s);
+    conf.setStringConverter(s -> s);
+
+    return conf;
+  }
+
+  public ConfigEntry<Integer> intConf() {
+    ConfigEntry<Integer> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    Function<String, Integer> func =
+        s -> {
+          if (s == null || s.isEmpty()) {
+            return null;
+          } else {
+            return Integer.parseInt(s);
+          }
+        };
+    conf.setValueConverter(func);
+
+    Function<Integer, String> stringFunc =
+        t -> Optional.ofNullable(t).map(String::valueOf).orElse(null);
+    conf.setStringConverter(stringFunc);
+
+    return conf;
+  }
+
+  public ConfigEntry<Long> longConf() {
+    ConfigEntry<Long> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    Function<String, Long> func =
+        s -> {
+          if (s == null || s.isEmpty()) {
+            return null;
+          } else {
+            return Long.parseLong(s);
+          }
+        };
+    conf.setValueConverter(func);
+
+    Function<Long, String> stringFunc =
+        t -> Optional.ofNullable(t).map(String::valueOf).orElse(null);
+    conf.setStringConverter(stringFunc);
+
+    return conf;
+  }
+
+  public ConfigEntry<Boolean> booleanConf() {
+    ConfigEntry<Boolean> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    Function<String, Boolean> func =
+        s -> {
+          if (s == null || s.isEmpty()) {
+            return null;
+          } else {
+            return Boolean.parseBoolean(s);
+          }
+        };
+    conf.setValueConverter(func);
+
+    Function<Boolean, String> stringFunc =
+        t -> Optional.ofNullable(t).map(String::valueOf).orElse(null);
+    conf.setStringConverter(stringFunc);
+
+    return conf;
+  }
+}

core/src/main/java/com/datastrato/graviton/config/ConfigEntry.java

@@ -0,0 +1,121 @@
+package com.datastrato.graviton.config;
+
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.function.Function;
+import lombok.Getter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigEntry<T> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ConfigEntry.class);
+
+  @Getter private String key;
+
+  @Getter private List<String> alternatives;
+
+  @Getter private T defaultValue;
+
+  private Function<String, T> valueConverter;
+
+  private Function<T, String> stringConverter;
+
+  @Getter private String doc;
+
+  @Getter private String version;
+
+  @Getter private boolean isPublic;
+
+  @Getter private boolean isDeprecated;
+
+  private boolean isOptional;
+
+  ConfigEntry(
+      String key,
+      String version,
+      String doc,
+      List<String> alternatives,
+      boolean isPublic,
+      boolean isDeprecated) {
+    this.key = key;
+    this.version = version;
+    this.doc = doc;
+    this.alternatives = alternatives;
+    this.isPublic = isPublic;
+    this.isDeprecated = isDeprecated;
+    this.isOptional = false;
+  }
+
+  void setValueConverter(Function<String, T> valueConverter) {
+    this.valueConverter = valueConverter;
+  }
+
+  void setStringConverter(Function<T, String> stringConverter) {
+    this.stringConverter = stringConverter;
+  }
+
+  void setDefaultValue(T t) {
+    this.defaultValue = t;
+  }
+
+  void setOptional() {
+    this.isOptional = true;
+  }
+
+  public ConfigEntry<T> createWithDefault(T t) {
+    ConfigEntry<T> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    conf.setValueConverter(valueConverter);
+    conf.setStringConverter(stringConverter);
+    conf.setDefaultValue(t);
+
+    return conf;
+  }
+
+  public ConfigEntry<Optional<T>> createWithOptional() {
+    ConfigEntry<Optional<T>> conf =
+        new ConfigEntry<>(key, version, doc, alternatives, isPublic, isDeprecated);
+    conf.setValueConverter(s -> Optional.ofNullable(valueConverter.apply(s)));
+    // null value should not be possible unless the user explicitly sets it
+    conf.setStringConverter(t -> t.map(stringConverter).orElse(null));
+    conf.setOptional();
+
+    return conf;
+  }
+
+  public T readFrom(Map<String, String> properties) throws NoSuchElementException {
+    String value = properties.get(key);
+    if (value == null) {
+      for (String alternative : alternatives) {
+        value = properties.get(alternative);
+        if (value != null) {
+          break;
+        }
+      }
+    }
+
+    if (value == null) {
+      if (defaultValue != null) {
+        return defaultValue;
+      } else if (!isOptional) {
+        throw new NoSuchElementException("No configuration found for key " + key);
+      }
+    }
+
+    return valueConverter.apply(value);
+  }
+
+  public void writeTo(Map<String, String> properties, T value) {
+    String stringValue = stringConverter.apply(value);
+    if (stringValue == null) {
+      // We don't want user to set a null value to config, so this basically will not happen;
+      LOG.warn("Config {} value to set is null, ignore setting to Config.", stringValue);
+      return;
+    }
+
+    properties.put(key, stringValue);
+  }
+}