From ba992ac00b164ff72050baa53ab93c0e68446a55 Mon Sep 17 00:00:00 2001
From: Fangbin Sun <sunfangbin@gmail.com>
Date: Mon, 24 Feb 2020 21:46:31 +0800
Subject: [PATCH] Fix broker to specify a list of bookie groups. (#6349)

### Motivation

Fixes #6343

### Modifications

Add a method to cast object value to `String`.
---
 .../ZkIsolatedBookieEnsemblePlacementPolicy.java | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkIsolatedBookieEnsemblePlacementPolicy.java b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkIsolatedBookieEnsemblePlacementPolicy.java
index 656b3a2029101..ef4f72bb14901 100644
--- a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkIsolatedBookieEnsemblePlacementPolicy.java
+++ b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkIsolatedBookieEnsemblePlacementPolicy.java
@@ -71,7 +71,7 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
             Optional<DNSToSwitchMapping> optionalDnsResolver, HashedWheelTimer timer, FeatureProvider featureProvider,
             StatsLogger statsLogger) {
         if (conf.getProperty(ISOLATION_BOOKIE_GROUPS) != null) {
-            String isolationGroupsString = (String) conf.getProperty(ISOLATION_BOOKIE_GROUPS);
+            String isolationGroupsString = castToString(conf.getProperty(ISOLATION_BOOKIE_GROUPS));
             if (!isolationGroupsString.isEmpty()) {
                 for (String isolationGroup : isolationGroupsString.split(",")) {
                     primaryIsolationGroups.add(isolationGroup);
@@ -80,7 +80,7 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
             }
         }
         if (conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS) != null) {
-            String secondaryIsolationGroupsString = (String) conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS);
+            String secondaryIsolationGroupsString = castToString(conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS));
             if (!secondaryIsolationGroupsString.isEmpty()) {
                 for (String isolationGroup : secondaryIsolationGroupsString.split(",")) {
                     secondaryIsolationGroups.add(isolationGroup);
@@ -90,6 +90,18 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
         return super.initialize(conf, optionalDnsResolver, timer, featureProvider, statsLogger);
     }
 
+    private String castToString(Object obj) {
+        if (obj instanceof List<?>) {
+            List<String> result = new ArrayList<>();
+            for (Object o : (List<?>) obj) {
+                result.add(String.class.cast(o));
+            }
+            return String.join(",", result);
+        } else {
+            return obj.toString();
+        }
+    }
+
     private ZooKeeperCache getAndSetZkCache(Configuration conf) {
         ZooKeeperCache zkCache = null;
         if (conf.getProperty(ZooKeeperCache.ZK_CACHE_INSTANCE) != null) {