[improve] [broker] Verify at least one URL is present for cluster creation

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/ClusterDataImpl.java

@@ -412,7 +412,16 @@ public ClusterDataImpl build() {
      *
      * @throws IllegalArgumentException exist illegal property.
      */
-    public void checkPropertiesIfPresent() throws IllegalArgumentException {
+     public void checkPropertiesIfPresent() throws IllegalArgumentException {
+
+        if (StringUtils.isEmpty(getServiceUrl()) && StringUtils.isEmpty(getServiceUrlTls())) {
+            throw new IllegalArgumentException("At least one of ServiceUrl or ServiceUrlTls must be set.");
+        }
+
+        if (StringUtils.isEmpty(getBrokerServiceUrl()) && StringUtils.isEmpty(getBrokerServiceUrlTls())) {
+            throw new IllegalArgumentException("At least one of BrokerServiceUrl or BrokerServiceUrlTls must be set.");
+        }
+
         URIPreconditions.checkURIIfPresent(getServiceUrl(),
                 uri -> Objects.equals(uri.getScheme(), "http"),
                 "Illegal service url, example: http://pulsar.example.com:8080");
@@ -430,20 +439,5 @@ public void checkPropertiesIfPresent() throws IllegalArgumentException {
                         || Objects.equals(uri.getScheme(), "pulsar+ssl"),
                 "Illegal proxy service url, example: pulsar+ssl://ats-proxy.example.com:4443 "
                         + "or pulsar://ats-proxy.example.com:4080");
-
-        warnIfUrlIsNotPresent();
-    }
-
-    private void warnIfUrlIsNotPresent() {
-        if (StringUtils.isEmpty(getServiceUrl()) && StringUtils.isEmpty(getServiceUrlTls())) {
-            log.warn("Service url not found, "
-                    + "please provide either service url, example: http://pulsar.example.com:8080 "
-                    + "or service tls url, example: https://pulsar.example.com:8443");
-        }
-        if (StringUtils.isEmpty(getBrokerServiceUrl()) && StringUtils.isEmpty(getBrokerServiceUrlTls())) {
-            log.warn("Broker service url not found, "
-                    + "please provide either broker service url, example: pulsar://pulsar.example.com:6650 "
-                    + "or broker service tls url, example: pulsar+ssl://pulsar.example.com:6651.");
-        }
     }
 }

pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/ClusterDataImplTest.java

@@ -20,6 +20,10 @@
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotSame;
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.jupiter.api.Test;
 
 import org.apache.pulsar.client.api.ProxyProtocol;
 import org.testng.annotations.Test;
@@ -62,4 +66,34 @@ public void verifyClone() {
         assertEquals(clone, originalData, "Clones should have object equality.");
         assertNotSame(clone, originalData, "Clones should not be the same reference.");
     }
+
+    @Test
+    public void testBothServiceUrlsisEmpty(){
+        ClusterDataImpl clusterData = new ClusterDataImpl();
+        clusterData.setServiceUrl("");
+        clusterData.setServiceUrlTls("");
+        clusterData.setBrokerServiceUrl("pulsar://pulsar.example.com:6650");
+        clusterData.setBrokerServiceUrlTls("pulsar+ssl://pulsar.example.com:6651");
+
+        IllegalArgumentException exception = assertThrows(
+            IllegalArgumentException.class, clusterData::checkPropertiesIfPresent
+        );
+
+        assertEquals("At least one of ServiceUrl or ServiceUrlTls must be set.", exception.getMessage());
+    }
+
+    @Test
+    public void testBothBrokerServiceUrlsisEmpty(){
+        ClusterDataImpl clusterData = new ClusterDataImpl();
+        clusterData.setServiceUrl("http://pulsar.example.com:8080");
+        clusterData.setServiceUrlTls("https://pulsar.example.com:8443");
+        clusterData.setBrokerServiceUrl("");
+        clusterData.setBrokerServiceUrlTls("");
+
+        IllegalArgumentException exception = assertThrows(
+            IllegalArgumentException.class, clusterData::checkPropertiesIfPresent
+        );
+
+        assertEquals("At least one of BrokerServiceUrl or BrokerServiceUrlTls must be set.", exception.getMessage());
+    }
 }