Validate rack name (apache#14336)

If the rack name set in the following case:
-  Enabled rack aware placement policy, and user set rack name which contains "/" in addition to the head and tail of the string. Such as "/r/a" or "r/a/b"
- Enabled region aware placement policy, and user set the rack name which contains multiple "/" in addition to the head and tail of the string. Such as "/region/region/a" or "region/a/rack/b"

(cherry picked from commit 25408f5)
(cherry picked from commit c7cf50a)

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Bookies.java

@@ -40,6 +40,7 @@
 import org.apache.bookkeeper.discover.RegistrationClient;
 import org.apache.bookkeeper.meta.MetadataClientDriver;
 import org.apache.bookkeeper.net.BookieId;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.common.policies.data.BookieInfo;
@@ -57,6 +58,7 @@
 @Api(value = "/bookies", description = "Configure bookies rack placement", tags = "bookies")
 @Produces(MediaType.APPLICATION_JSON)
 public class Bookies extends AdminResource {
+    private static final String PATH_SEPARATOR = "/";
 
     @GET
     @Path("/racks-info")
@@ -156,6 +158,19 @@ public void updateBookieRackInfo(@PathParam("bookie") String bookieAddress, @Que
             throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'group' parameters is missing");
         }
 
+        // validate rack name
+        int separatorCnt = StringUtils.countMatches(
+                StringUtils.strip(bookieInfo.getRack(), PATH_SEPARATOR), PATH_SEPARATOR);
+        boolean isRackEnabled = pulsar().getConfiguration().isBookkeeperClientRackawarePolicyEnabled();
+        boolean isRegionEnabled = pulsar().getConfiguration().isBookkeeperClientRegionawarePolicyEnabled();
+        if (isRackEnabled && ((isRegionEnabled && separatorCnt != 1) || (!isRegionEnabled && separatorCnt != 0))) {
+            throw new RestException(Status.PRECONDITION_FAILED, "Bookie 'rack' parameter is invalid, "
+                    + "When `RackawareEnsemblePlacementPolicy` is enabled, the rack name is not allowed to contain "
+                    + "slash (`/`) except for the beginning and end of the rack name string. "
+                    + "When `RegionawareEnsemblePlacementPolicy` is enabled, the rack name can only contain "
+                    + "one slash (`/`) except for the beginning and end of the rack name string.");
+        }
+
         Optional<Entry<BookiesRackConfiguration, Stat>> entry = localZkCache()
                 .getEntry(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, (key, content) -> ObjectMapperFactory
                         .getThreadLocal().readValue(content, BookiesRackConfiguration.class));

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/BookiesApiTest.java

@@ -18,13 +18,14 @@
  */
 package org.apache.pulsar.broker.admin;
 
+import static org.mockito.Mockito.doReturn;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.bookkeeper.client.PulsarMockBookKeeper;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.policies.data.BookieInfo;
@@ -123,6 +124,61 @@ public void testBasic() throws Exception {
                 .get()
                 .getValue()
                 .size());
+
+        // test invalid rack name
+        // use rack aware placement policy
+        String errorMsg = "Bookie 'rack' parameter is invalid, When `RackawareEnsemblePlacementPolicy` is enabled, "
+            + "the rack name is not allowed to contain slash (`/`) except for the beginning and end of the rack name "
+            + "string. When `RegionawareEnsemblePlacementPolicy` is enabled, the rack name can only contain "
+            + "one slash (`/`) except for the beginning and end of the rack name string.";
+
+        BookieInfo newInfo3 = BookieInfo.builder()
+            .rack("/rack/a")
+            .hostname("127.0.0.2")
+            .build();
+        try {
+            admin.bookies().updateBookieRackInfo(bookie0, "default", newInfo3);
+            fail();
+        } catch (PulsarAdminException e) {
+            assertEquals(412, e.getStatusCode());
+            assertEquals(errorMsg, e.getMessage());
+        }
+
+        BookieInfo newInfo4 = BookieInfo.builder()
+            .rack("/rack")
+            .hostname("127.0.0.2")
+            .build();
+        try {
+            admin.bookies().updateBookieRackInfo(bookie0, "default", newInfo4);
+        } catch (PulsarAdminException e) {
+            fail();
+        }
+
+        // enable region aware placement policy
+        ServiceConfiguration configuration = new ServiceConfiguration();
+        configuration.setBookkeeperClientRegionawarePolicyEnabled(true);
+        doReturn(configuration).when(pulsar).getConfiguration();
+        BookieInfo newInfo5 = BookieInfo.builder()
+            .rack("/region/rack/a")
+            .hostname("127.0.0.2")
+            .build();
+        try {
+            admin.bookies().updateBookieRackInfo(bookie0, "default", newInfo5);
+            fail();
+        } catch (PulsarAdminException e) {
+            assertEquals(412, e.getStatusCode());
+            assertEquals(errorMsg, e.getMessage());
+        }
+
+        BookieInfo newInfo6 = BookieInfo.builder()
+            .rack("/region/rack/")
+            .hostname("127.0.0.2")
+            .build();
+        try {
+            admin.bookies().updateBookieRackInfo(bookie0, "default", newInfo6);
+        } catch (PulsarAdminException e) {
+            fail();
+        }
     }
 
 }