private void testConnectServiceUrl(String serviceUrl) {
        // test connect for Pulsar ServiceUrl
        Preconditions.expectNotNull(serviceUrl, "Pulsar ServiceUrl is empty");
        Preconditions.expectTrue(serviceUrl.startsWith(SERVICE_URL_PREFIX),
                String.format("Pulsar ServiceUrl=%s should starts with %s", serviceUrl, SERVICE_URL_PREFIX));

        String hostPortStr = serviceUrl.replaceAll(SERVICE_URL_PREFIX, "");

        boolean successConnect = Arrays.stream(hostPortStr.split(InlongConstants.COMMA))
                // Only when all addresses are reachable, it's considered successful
                .allMatch(hostPort -> {
                    String[] hostPortArr = hostPort.split(InlongConstants.COLON);
                    Preconditions.expectTrue(hostPortArr.length >= 2,
                            String.format("Pulsar ServiceUrl=%s should has ip and port, such as '127.0.0.1:6650'",
                                    serviceUrl));

                    String host = hostPortArr[0];
                    int port = Integer.parseInt(hostPortArr[1]);

                    try (Socket socket = new Socket()) {
                        SocketAddress socketAddress = new InetSocketAddress(host, port);
                        socket.connect(socketAddress, 30000);
                        return true;
                    } catch (IOException e) {
                        String errMsg = String.format("Pulsar connection failed for ServiceUrl=%s", hostPort);
                        LOGGER.error(errMsg, e);
                        return false;
                    }
                });

        Preconditions.expectTrue(successConnect,
                String.format("Pulsar connection failed for ServiceUrl=%s", hostPortStr));
    }
Test connect for Pulsar ServiceUrl

@param serviceUrl Pulsar serviceUrl, such as: pulsar://127.0.0.1:6650,127.0.0.2:6650
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/PulsarClusterOperator.java