[IMPROVEMENT] Introduce the enumType in ConfigOptions

common/src/main/java/org/apache/uniffle/common/config/ConfigOptions.java

@@ -139,6 +139,15 @@ public TypedConfigOptionBuilder<Double> doubleType() {
     public TypedConfigOptionBuilder<String> stringType() {
       return new TypedConfigOptionBuilder<>(key, String.class);
     }
+
+    /**
+     * Defines that the value of the option should be of {@link Enum} type.
+     *
+     * @param enumClass Concrete type of the expected enum.
+     */
+    public <T extends Enum<T>> TypedConfigOptionBuilder<T> enumType(Class<T> enumClass) {
+      return new TypedConfigOptionBuilder<>(key, enumClass);
+    }
   }
 
   // ------------------------------------------------------------------------

common/src/main/java/org/apache/uniffle/common/config/ConfigUtils.java

@@ -18,7 +18,9 @@
 package org.apache.uniffle.common.config;
 
 import java.lang.reflect.Field;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.function.Function;
 
 import com.google.common.collect.Lists;
@@ -63,10 +65,32 @@ public static <T> T convertValue(Object rawValue, Class<?> clazz) {
       return (T) convertToDouble(rawValue);
     } else if (String.class.equals(clazz)) {
       return (T) convertToString(rawValue);
+    } else if (clazz.isEnum()) {
+      return (T) convertToEnum(rawValue, (Class<? extends Enum<?>>) clazz);
     }
     throw new IllegalArgumentException("Unsupported type: " + clazz);
   }
 
+  public static <E extends Enum<?>> E convertToEnum(Object o, Class<E> clazz) {
+    if (o.getClass().equals(clazz)) {
+      return (E) o;
+    }
+
+    return Arrays.stream(clazz.getEnumConstants())
+        .filter(
+            e ->
+                e.toString()
+                    .toUpperCase(Locale.ROOT)
+                    .equals(o.toString().toUpperCase(Locale.ROOT)))
+        .findAny()
+        .orElseThrow(
+            () ->
+                new IllegalArgumentException(
+                    String.format(
+                        "Could not parse value for enum %s. Expected one of: [%s]",
+                        clazz, Arrays.toString(clazz.getEnumConstants()))));
+  }
+
   static String convertToString(Object o) {
     if (o.getClass() == String.class) {
       return (String) o;

common/src/test/java/org/apache/uniffle/common/config/ConfigOptionTest.java

@@ -56,6 +56,34 @@ public void testSetKVWithStringTypeDirectly() {
     assertFalse(conf.get(booleanConfig));
   }
 
+  enum TestType {
+    TYPE_1,
+    TYPE_2,
+  }
+
+  @Test
+  public void testEnumType() {
+    final ConfigOption<TestType> enumConfigOption = ConfigOptions
+        .key("rss.enum")
+        .enumType(TestType.class)
+        .defaultValue(TestType.TYPE_1)
+        .withDescription("enum test");
+
+    RssBaseConf conf = new RssBaseConf();
+
+    // case1: default value
+    assertEquals(TestType.TYPE_1, conf.get(enumConfigOption));
+
+    // case2: return the user specified value
+    conf.set(enumConfigOption, TestType.TYPE_2);
+    assertEquals(TestType.TYPE_2, conf.get(enumConfigOption));
+
+    // case3: set enum val with string
+    conf = new RssBaseConf();
+    conf.setString("rss.enum", "TYPE_2");
+    assertEquals(TestType.TYPE_2, conf.get(enumConfigOption));
+  }
+
   @Test
   public void testListTypes() {
     // test the string type list.

coordinator/src/main/java/org/apache/uniffle/coordinator/AssignmentStrategyFactory.java

@@ -28,17 +28,17 @@ public AssignmentStrategyFactory(CoordinatorConf conf, ClusterManager clusterMan
   }
 
   public AssignmentStrategy getAssignmentStrategy() {
-    String strategy = conf.getString(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY);
-    if (StrategyName.BASIC.name().equals(strategy)) {
+    StrategyName strategy = conf.get(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY);
+    if (StrategyName.BASIC == strategy) {
       return new BasicAssignmentStrategy(clusterManager);
-    } else if (StrategyName.PARTITION_BALANCE.name().equals(strategy)) {
+    } else if (StrategyName.PARTITION_BALANCE == strategy) {
       return new PartitionBalanceAssignmentStrategy(clusterManager);
     } else {
       throw new UnsupportedOperationException("Unsupported assignment strategy.");
     }
   }
 
-  private enum StrategyName {
+  public enum StrategyName {
     BASIC,
     PARTITION_BALANCE
   }

coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorConf.java

@@ -26,6 +26,8 @@
 import org.apache.uniffle.common.config.RssBaseConf;
 import org.apache.uniffle.common.util.RssUtils;
 
+import static org.apache.uniffle.coordinator.AssignmentStrategyFactory.StrategyName.PARTITION_BALANCE;
+
 /**
  * Configuration for Coordinator Service and rss-cluster, including service port,
  * heartbeat interval and etc.
@@ -54,10 +56,11 @@ public class CoordinatorConf extends RssBaseConf {
       .defaultValue(30L)
       .withDescription("The periodic interval times of output alive nodes. The interval sec can be calculated by ("
           + COORDINATOR_HEARTBEAT_TIMEOUT.key() + "/3 * rss.coordinator.server.periodic.output.interval.times)");
-  public static final ConfigOption<String> COORDINATOR_ASSIGNMENT_STRATEGY = ConfigOptions
+  public static final ConfigOption<AssignmentStrategyFactory.StrategyName>
+      COORDINATOR_ASSIGNMENT_STRATEGY = ConfigOptions
       .key("rss.coordinator.assignment.strategy")
-      .stringType()
-      .defaultValue("PARTITION_BALANCE")
+      .enumType(AssignmentStrategyFactory.StrategyName.class)
+      .defaultValue(PARTITION_BALANCE)
       .withDescription("Strategy for assigning shuffle server to write partitions");
   public static final ConfigOption<Long> COORDINATOR_APP_EXPIRED = ConfigOptions
       .key("rss.coordinator.app.expired")

coordinator/src/test/java/org/apache/uniffle/coordinator/CoordinatorConfTest.java

@@ -42,7 +42,7 @@ public void test() {
     assertEquals(123, conf.getLong(CoordinatorConf.COORDINATOR_HEARTBEAT_TIMEOUT));
 
     // test default conf
-    assertEquals("PARTITION_BALANCE", conf.getString(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY));
+    assertEquals("PARTITION_BALANCE", conf.get(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY).name());
     assertEquals(256, conf.getInteger(CoordinatorConf.JETTY_CORE_POOL_SIZE));
     assertEquals(60 * 1000, conf.getLong(CoordinatorConf.COORDINATOR_EXCLUDE_NODES_CHECK_INTERVAL));
 

integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcTest.java

@@ -60,7 +60,7 @@ public class CoordinatorGrpcTest extends CoordinatorTestBase {
   public static void setupServers() throws Exception {
     CoordinatorConf coordinatorConf = getCoordinatorConf();
     coordinatorConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
-    coordinatorConf.set(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY, "BASIC");
+    coordinatorConf.setString(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY.key(), "BASIC");
     coordinatorConf.setLong("rss.coordinator.app.expired", 2000);
     coordinatorConf.setLong("rss.coordinator.server.heartbeat.timeout", 3000);
     createCoordinatorServer(coordinatorConf);