Fix NPE when reading ZK address from controller config (#9751)

pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java

@@ -32,7 +32,6 @@
 import org.apache.pinot.spi.env.PinotConfiguration;
 import org.apache.pinot.spi.filesystem.LocalPinotFS;
 import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.StringUtil;
 import org.apache.pinot.spi.utils.TimeUtils;
 
 import static org.apache.pinot.spi.utils.CommonConstants.Controller.CONFIG_OF_CONTROLLER_METRICS_PREFIX;
@@ -304,8 +303,7 @@ public void setPinotFSFactoryClasses(Configuration pinotFSFactoryClasses) {
     setProperty(PINOT_FS_FACTORY_CLASS_LOCAL, DEFAULT_PINOT_FS_FACTORY_CLASS_LOCAL);
 
     if (pinotFSFactoryClasses != null) {
-      pinotFSFactoryClasses.getKeys()
-          .forEachRemaining(key -> setProperty((String) key, pinotFSFactoryClasses.getProperty((String) key)));
+      pinotFSFactoryClasses.getKeys().forEachRemaining(key -> setProperty(key, pinotFSFactoryClasses.getProperty(key)));
     }
   }
 
@@ -437,21 +435,11 @@ public String generateVipUrl() {
   }
 
   public String getZkStr() {
-    Object zkAddressObj = containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER) ? getProperty(
+    String zkAddress = containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER) ? getProperty(
         CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER) : getProperty(ZK_STR);
-
-    // The set method converted comma separated string into ArrayList, so need to convert back to String here.
-    if (zkAddressObj instanceof List) {
-      List<String> zkAddressList = (List<String>) zkAddressObj;
-      String[] zkAddress = zkAddressList.toArray(new String[0]);
-      return StringUtil.join(",", zkAddress);
-    } else if (zkAddressObj instanceof String) {
-      return (String) zkAddressObj;
-    } else {
-      throw new RuntimeException(
-          "Unexpected data type for zkAddress PropertiesConfiguration, expecting String but got " + zkAddressObj
-              .getClass().getName());
-    }
+    Preconditions.checkState(zkAddress != null,
+        "ZK address is not configured. Please configure it using the config: 'pinot.zk.server'");
+    return zkAddress;
   }
 
   @Override
@@ -523,8 +511,8 @@ public void setRetentionControllerFrequencyInSeconds(int retentionFrequencyInSec
    * @return the supplied config in seconds
    */
   public int getOfflineSegmentIntervalCheckerFrequencyInSeconds() {
-    return Optional
-        .ofNullable(getProperty(ControllerPeriodicTasksConf.OFFLINE_SEGMENT_INTERVAL_CHECKER_FREQUENCY_PERIOD))
+    return Optional.ofNullable(
+            getProperty(ControllerPeriodicTasksConf.OFFLINE_SEGMENT_INTERVAL_CHECKER_FREQUENCY_PERIOD))
         .map(period -> (int) convertPeriodToSeconds(period)).orElseGet(() -> getProperty(
             ControllerPeriodicTasksConf.DEPRECATED_OFFLINE_SEGMENT_INTERVAL_CHECKER_FREQUENCY_IN_SECONDS,
             ControllerPeriodicTasksConf.DEFAULT_OFFLINE_SEGMENT_INTERVAL_CHECKER_FREQUENCY_IN_SECONDS));
@@ -781,7 +769,6 @@ public String getAccessControlFactoryClass() {
     return getProperty(ACCESS_CONTROL_FACTORY_CLASS, DEFAULT_ACCESS_CONTROL_FACTORY_CLASS);
   }
 
-
   public void setAccessControlFactoryClass(String accessControlFactoryClass) {
     setProperty(ACCESS_CONTROL_FACTORY_CLASS, accessControlFactoryClass);
   }
@@ -882,7 +869,7 @@ public void setControllerMode(ControllerMode controllerMode) {
   }
 
   public ControllerMode getControllerMode() {
-    return ControllerMode.valueOf(getProperty(CONTROLLER_MODE, DEFAULT_CONTROLLER_MODE.toString()).toUpperCase());
+    return ControllerMode.valueOf(getProperty(CONTROLLER_MODE, DEFAULT_CONTROLLER_MODE).toUpperCase());
   }
 
   public void setLeadControllerResourceRebalanceStrategy(String rebalanceStrategy) {
@@ -911,8 +898,8 @@ public int getControllerBrokerPortOverride() {
   }
 
   public List<String> getTableConfigTunerPackages() {
-    return Arrays
-        .asList(getProperty(TABLE_CONFIG_TUNER_PACKAGES, DEFAULT_TABLE_CONFIG_TUNER_PACKAGES).split("\\s*,\\s*"));
+    return Arrays.asList(
+        getProperty(TABLE_CONFIG_TUNER_PACKAGES, DEFAULT_TABLE_CONFIG_TUNER_PACKAGES).split("\\s*,\\s*"));
   }
 
   public String getControllerResourcePackages() {