diff --git a/common/src/main/java/com/tencent/rss/common/config/ConfigOptions.java b/common/src/main/java/com/tencent/rss/common/config/ConfigOptions.java
index 2299208e67..d7edf6de0b 100644
--- a/common/src/main/java/com/tencent/rss/common/config/ConfigOptions.java
+++ b/common/src/main/java/com/tencent/rss/common/config/ConfigOptions.java
@@ -18,8 +18,11 @@
 
 package com.tencent.rss.common.config;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * {@code ConfigOptions} are used to build a {@link ConfigOption}.
@@ -54,6 +57,7 @@
  * }</pre>
  */
 public class ConfigOptions {
+
   /**
    * Not intended to be instantiated.
    */
@@ -167,6 +171,10 @@ public static class TypedConfigOptionBuilder<T> {
       this.converter = converter;
     }
 
+    public ListConfigOptionBuilder asList() {
+      return new ListConfigOptionBuilder<T>(key, clazz, converter);
+    }
+
     // todo: errorMsg shouldn't contain key
     public TypedConfigOptionBuilder checkValue(Function<T, Boolean> checkValue, String errMsg) {
       Function<Object, T> newConverter = (v) -> {
@@ -208,4 +216,75 @@ public ConfigOption<T> noDefaultValue() {
         converter);
     }
   }
+
+  /**
+   * Builder for {@link ConfigOption} of list of type {@link E}.
+   *
+   * @param <E> list element type of the option
+   */
+  public static class ListConfigOptionBuilder<E> {
+    private static final String LIST_SPILTTER = ",";
+
+    private final String key;
+    private final Class<E> clazz;
+    private final Function<Object, E> atomicConverter;
+    private Function<Object, List<E>> asListConverter;
+
+    public ListConfigOptionBuilder(String key, Class<E> clazz, Function<Object, E> atomicConverter) {
+      this.key = key;
+      this.clazz = clazz;
+      this.atomicConverter = atomicConverter;
+      this.asListConverter = (v) -> {
+        if (v instanceof List) {
+          return (List<E>) v;
+        } else {
+          return Arrays.stream(v.toString().split(LIST_SPILTTER))
+                  .map(s -> atomicConverter.apply(s)).collect(Collectors.toList());
+        }
+      };
+    }
+
+    public ListConfigOptionBuilder checkValue(Function<E, Boolean> checkValueFunc, String errMsg) {
+      final Function<Object, List<E>> listConverFunc = asListConverter;
+      Function<Object, List<E>> newConverter = (v) -> {
+        List<E> list = listConverFunc.apply(v);
+        if (list.stream().anyMatch(x -> !checkValueFunc.apply(x))) {
+          throw new IllegalArgumentException(errMsg);
+        }
+        return list;
+      };
+      this.asListConverter = newConverter;
+      return this;
+    }
+
+    /**
+     * Creates a ConfigOption with the given default value.
+     *
+     * @param values The list of default values for the config option
+     * @return The config option with the default value.
+     */
+    @SafeVarargs
+    public final ConfigOption<List<E>> defaultValues(E... values) {
+      return new ConfigOption<>(
+        key,
+        clazz,
+        ConfigOption.EMPTY_DESCRIPTION,
+        Arrays.asList(values),
+        asListConverter);
+    }
+
+    /**
+     * Creates a ConfigOption without a default value.
+     *
+     * @return The config option without a default value.
+     */
+    public ConfigOption<List<E>> noDefaultValue() {
+      return new ConfigOption<>(
+        key,
+        clazz,
+        ConfigOption.EMPTY_DESCRIPTION,
+        null,
+        asListConverter);
+    }
+  }
 }
diff --git a/common/src/test/java/com/tencent/rss/common/config/ConfigOptionTest.java b/common/src/test/java/com/tencent/rss/common/config/ConfigOptionTest.java
index f693927c28..f413872b3c 100644
--- a/common/src/test/java/com/tencent/rss/common/config/ConfigOptionTest.java
+++ b/common/src/test/java/com/tencent/rss/common/config/ConfigOptionTest.java
@@ -22,12 +22,118 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
+import java.util.List;
+import java.util.function.Function;
+
+import com.google.common.collect.Lists;
+
 public class ConfigOptionTest {
 
+  @Test
+  public void testSetKVWithStringTypeDirectly() {
+    final ConfigOption<Integer> intConfig = ConfigOptions
+            .key("rss.key1")
+            .intType()
+            .defaultValue(1000)
+            .withDescription("Int config key1");
+
+    RssConf conf = new RssBaseConf();
+    conf.setString("rss.key1", "2000");
+    assertEquals(2000, conf.get(intConfig));
+
+    final ConfigOption<Boolean> booleanConfig = ConfigOptions
+            .key("key2")
+            .booleanType()
+            .defaultValue(false)
+            .withDescription("Boolean config key");
+
+    conf.setString("key2", "true");
+    assertTrue(conf.get(booleanConfig));
+    conf.setString("key2", "False");
+    assertFalse(conf.get(booleanConfig));
+  }
+
+  @Test
+  public void testListTypes() {
+    // test the string type list.
+    final ConfigOption<List<String>> listStringConfigOption = ConfigOptions
+            .key("rss.key1")
+            .stringType()
+            .asList()
+            .defaultValues("h1", "h2")
+            .withDescription("List config key1");
+
+    List<String> defaultValues = listStringConfigOption.defaultValue();
+    assertEquals(2, defaultValues.size());
+    assertSame(String.class, listStringConfigOption.getClazz());
+
+    RssBaseConf conf = new RssBaseConf();
+    conf.set(listStringConfigOption, Lists.newArrayList("a", "b", "c"));
+
+    List<String> vals = conf.get(listStringConfigOption);
+    assertEquals(3, vals.size());
+    assertEquals(Lists.newArrayList("a", "b", "c"), vals);
+
+    // test the long type list
+    final ConfigOption<List<Long>> listLongConfigOption = ConfigOptions
+            .key("rss.key2")
+            .longType()
+            .asList()
+            .defaultValues(1L)
+            .withDescription("List long config key2");
+
+    List<Long> longDefaultVals = listLongConfigOption.defaultValue();
+    assertEquals(longDefaultVals.size(), 1);
+    assertEquals(Lists.newArrayList(1L), longDefaultVals);
+
+    conf.setString("rss.key2", "1,2,3");
+    List<Long> longVals = conf.get(listLongConfigOption);
+    assertEquals(Lists.newArrayList(1L, 2L, 3L), longVals);
+
+    // test overwrite the same conf key.
+    conf.set(listLongConfigOption, Lists.newArrayList(1L, 2L, 3L, 4L));
+    assertEquals(Lists.newArrayList(1L, 2L, 3L, 4L), conf.get(listLongConfigOption));
+
+    // test the no-default values
+    final ConfigOption<List<Long>> listLongConfigOptionWithoutDefault = ConfigOptions
+            .key("rss.key3")
+            .longType()
+            .asList()
+            .noDefaultValue()
+            .withDescription("List long config key3 without default values");
+    List<Long> valsWithoutDefault = listLongConfigOptionWithoutDefault.defaultValue();
+    assertNull(valsWithoutDefault);
+
+    // test the method of check
+    final ConfigOption<List<Integer>> checkLongValsOptions = ConfigOptions
+            .key("rss.key4")
+            .intType()
+            .asList()
+            .checkValue((Function<Integer, Boolean>) val -> val > 0, "Every number of list should be positive")
+            .noDefaultValue()
+            .withDescription("The key4 is illegal");
+
+    conf.set(checkLongValsOptions, Lists.newArrayList(-1, 2, 3));
+
+    try {
+      conf.get(checkLongValsOptions);
+      fail();
+    } catch (IllegalArgumentException illegalArgumentException) {
+    }
+
+    conf.set(checkLongValsOptions, Lists.newArrayList(1, 2, 3));
+    try {
+      conf.get(checkLongValsOptions);
+    } catch (IllegalArgumentException illegalArgumentException) {
+      fail();
+    }
+  }
+
   @Test
   public void testBasicTypes() {
     final ConfigOption<Integer> intConfig = ConfigOptions