[fix][bookie] Correctly handle list configuration values (#17661)

* [fix][bookie] Correctly handle list configuration values

* Remove unused import

### Motivation

When the `metadataServiceUri` or the `zkServers` configuration for `BookieRackAffinityMapping` is provided as a comma delimited list, it is automatically parsed into an ArrayList by the configuration class because the Bookkeeper configuration class relies on the defaults in the `org.apache.commons.configuration.AbstractConfiguration` class. Here is a sample error:

```
2022-07-29T19:25:43,437+0000 [main] ERROR org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicyImpl - Failed to initialize DNS Resolver org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping, used default subnet resolver
java.lang.ClassCastException: class java.util.ArrayList cannot be cast to class java.lang.String (java.util.ArrayList and java.lang.String are in module java.base of loader 'bootstrap')
```

Also, see #6349 and #6343 for context.

### Modifications

* Move the `castToString` method out to a shared class.
* Use the `castToString` method to safely get the configuration value.

### Verifying this change

This PR includes a test.

### Documentation

- [x] `doc-not-needed`

Author: michaeljmarshall

pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMapping.java

@@ -77,7 +77,7 @@ public static MetadataStore createMetadataStore(Configuration conf) throws Metad
             store = (MetadataStore) storeProperty;
         } else {
             String url;
-            String metadataServiceUri = (String) conf.getProperty("metadataServiceUri");
+            String metadataServiceUri = ConfigurationStringUtil.castToString(conf.getProperty("metadataServiceUri"));
             if (StringUtils.isNotBlank(metadataServiceUri)) {
                 try {
                     url = metadataServiceUri.replaceFirst(METADATA_STORE_SCHEME + ":", "")
@@ -86,7 +86,7 @@ public static MetadataStore createMetadataStore(Configuration conf) throws Metad
                     throw new MetadataException(Code.METADATA_SERVICE_ERROR, e);
                 }
             } else {
-                String zkServers = (String) conf.getProperty("zkServers");
+                String zkServers = ConfigurationStringUtil.castToString(conf.getProperty("zkServers"));
                 if (StringUtils.isBlank(zkServers)) {
                     String errorMsg = String.format("Neither %s configuration set in the BK client configuration nor "
                             + "metadataServiceUri/zkServers set in bk server configuration", METADATA_STORE_INSTANCE);

pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/ConfigurationStringUtil.java

@@ -0,0 +1,48 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.bookie.rackawareness;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ConfigurationStringUtil {
+
+    /**
+     * The Bookkeeper configuration class converts comma delimited strings to ArrayLists, by default. Use
+     * this method to ensure a configuration value is a {@link String}.
+     * @param obj - object to convert to a string
+     * @return The object's conversion to a string where Lists map to a comma delimited list.
+     */
+    static String castToString(Object obj) {
+        if (null == obj) {
+            return "";
+        }
+        if (obj instanceof List<?>) {
+            List<String> result = new ArrayList<>();
+            for (Object o : (List<?>) obj) {
+                result.add((String) o);
+            }
+            return String.join(",", result);
+        } else {
+            return obj.toString();
+        }
+    }
+
+}

pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicy.java

@@ -20,7 +20,6 @@
 
 import static org.apache.pulsar.bookie.rackawareness.BookieRackAffinityMapping.METADATA_STORE_INSTANCE;
 import io.netty.util.HashedWheelTimer;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
@@ -80,7 +79,8 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
         Set<String> primaryIsolationGroups = new HashSet<>();
         Set<String> secondaryIsolationGroups = new HashSet<>();
         if (conf.getProperty(ISOLATION_BOOKIE_GROUPS) != null) {
-            String isolationGroupsString = castToString(conf.getProperty(ISOLATION_BOOKIE_GROUPS));
+            String isolationGroupsString = ConfigurationStringUtil
+                    .castToString(conf.getProperty(ISOLATION_BOOKIE_GROUPS));
             if (!isolationGroupsString.isEmpty()) {
                 for (String isolationGroup : isolationGroupsString.split(",")) {
                     primaryIsolationGroups.add(isolationGroup);
@@ -91,7 +91,8 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
             bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).join();
         }
         if (conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS) != null) {
-            String secondaryIsolationGroupsString = castToString(conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS));
+            String secondaryIsolationGroupsString = ConfigurationStringUtil
+                    .castToString(conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS));
             if (!secondaryIsolationGroupsString.isEmpty()) {
                 for (String isolationGroup : secondaryIsolationGroupsString.split(",")) {
                     secondaryIsolationGroups.add(isolationGroup);
@@ -102,21 +103,6 @@ public RackawareEnsemblePlacementPolicyImpl initialize(ClientConfiguration conf,
         return super.initialize(conf, optionalDnsResolver, timer, featureProvider, statsLogger, bookieAddressResolver);
     }
 
-    private static String castToString(Object obj) {
-        if (null == obj) {
-            return "";
-        }
-        if (obj instanceof List<?>) {
-            List<String> result = new ArrayList<>();
-            for (Object o : (List<?>) obj) {
-                result.add((String) o);
-            }
-            return String.join(",", result);
-        } else {
-            return obj.toString();
-        }
-    }
-
     @Override
     public PlacementResult<List<BookieId>> newEnsemble(int ensembleSize, int writeQuorumSize, int ackQuorumSize,
             Map<String, byte[]> customMetadata, Set<BookieId> excludeBookies)
@@ -181,9 +167,10 @@ private static Pair<Set<String>, Set<String>> getIsolationGroup(
         String className = IsolatedBookieEnsemblePlacementPolicy.class.getName();
         if (ensemblePlacementPolicyConfig.getPolicyClass().getName().equals(className)) {
             Map<String, Object> properties = ensemblePlacementPolicyConfig.getProperties();
-            String primaryIsolationGroupString = castToString(properties.getOrDefault(ISOLATION_BOOKIE_GROUPS, ""));
-            String secondaryIsolationGroupString =
-                    castToString(properties.getOrDefault(SECONDARY_ISOLATION_BOOKIE_GROUPS, ""));
+            String primaryIsolationGroupString = ConfigurationStringUtil
+                    .castToString(properties.getOrDefault(ISOLATION_BOOKIE_GROUPS, ""));
+            String secondaryIsolationGroupString = ConfigurationStringUtil
+                    .castToString(properties.getOrDefault(SECONDARY_ISOLATION_BOOKIE_GROUPS, ""));
             if (!primaryIsolationGroupString.isEmpty()) {
                 pair.setLeft(new HashSet(Arrays.asList(primaryIsolationGroupString.split(","))));
             } else {

pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMappingTest.java

@@ -82,6 +82,18 @@ public void testBasic() throws Exception {
         assertNull(racks.get(2));
     }
 
+    @Test
+    public void testMultipleMetadataServiceUris() {
+        BookieRackAffinityMapping mapping1 = new BookieRackAffinityMapping();
+        ClientConfiguration bkClientConf1 = new ClientConfiguration();
+        bkClientConf1.setProperty("metadataServiceUri", "memory:local,memory:local");
+        bkClientConf1.setProperty("zkTimeout", "100000");
+
+        mapping1.setBookieAddressResolver(BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+        // This previously threw an exception when the metadataServiceUri was a comma delimited list.
+        mapping1.setConf(bkClientConf1);
+    }
+
     @Test
     public void testInvalidRackName() {
         String data = "{\"group1\": {\"" + BOOKIE1