public static PulsarClient createClient(ServiceConfiguration brokerConfig) throws PulsarClientException {
        ClientBuilder clientBuilder = PulsarClient.builder()
                .memoryLimit(0, SizeUnit.BYTES);

        // Apply all arbitrary configuration. This must be called before setting any fields annotated as
        // @Secret on the ClientConfigurationData object because of the way they are serialized.
        // See https://github.com/apache/pulsar/issues/8509 for more information.
        clientBuilder.loadConf(PropertiesUtils.filterAndMapProperties(brokerConfig.getProperties(), "brokerClient_"));

        if (isNotBlank(brokerConfig.getBrokerClientAuthenticationPlugin())) {
            clientBuilder.authentication(brokerConfig.getBrokerClientAuthenticationPlugin(),
                    brokerConfig.getBrokerClientAuthenticationParameters());
        }

        AdvertisedListener internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar+ssl");
        if (internalListener.getBrokerServiceUrlTls() != null && brokerConfig.isBrokerClientTlsEnabled()) {
            clientBuilder.serviceUrl(internalListener.getBrokerServiceUrlTls().toString())
                    .allowTlsInsecureConnection(brokerConfig.isTlsAllowInsecureConnection())
                    .enableTlsHostnameVerification(brokerConfig.isTlsHostnameVerificationEnabled())
                    .sslFactoryPlugin(brokerConfig.getBrokerClientSslFactoryPlugin())
                    .sslFactoryPluginParams(brokerConfig.getBrokerClientSslFactoryPluginParams());
            if (brokerConfig.isBrokerClientTlsEnabledWithKeyStore()) {
                clientBuilder.useKeyStoreTls(true)
                        .tlsKeyStoreType(brokerConfig.getBrokerClientTlsKeyStoreType())
                        .tlsKeyStorePath(brokerConfig.getBrokerClientTlsKeyStore())
                        .tlsKeyStorePassword(brokerConfig.getBrokerClientTlsKeyStorePassword())
                        .tlsTrustStoreType(brokerConfig.getBrokerClientTlsTrustStoreType())
                        .tlsTrustStorePath(brokerConfig.getBrokerClientTlsTrustStore())
                        .tlsTrustStorePassword(brokerConfig.getBrokerClientTlsTrustStorePassword());
            } else {
                clientBuilder.tlsTrustCertsFilePath(brokerConfig.getBrokerClientTrustCertsFilePath())
                        .tlsKeyFilePath(brokerConfig.getBrokerClientKeyFilePath())
                        .tlsCertificateFilePath(brokerConfig.getBrokerClientCertificateFilePath());
            }
        } else {
            internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar");
            clientBuilder.serviceUrl(internalListener.getBrokerServiceUrl().toString());
        }

        return clientBuilder.build();
    }
@Test
    public void testUseTlsUrlWithPEM() throws PulsarClientException {
        ServiceConfiguration serviceConfiguration = spy(ServiceConfiguration.class);
        serviceConfiguration.setBrokerServicePortTls(Optional.of(6651));
        serviceConfiguration.setBrokerClientTlsEnabled(true);
        serviceConfiguration.setProperties(new Properties());

        @Cleanup
        PulsarClient ignored = CompactorTool.createClient(serviceConfiguration);

        verify(serviceConfiguration, times(1)).isBrokerClientTlsEnabled();
        verify(serviceConfiguration, times(1)).isTlsAllowInsecureConnection();
        verify(serviceConfiguration, times(1)).getBrokerClientKeyFilePath();
        verify(serviceConfiguration, times(1)).getBrokerClientTrustCertsFilePath();
        verify(serviceConfiguration, times(1)).getBrokerClientCertificateFilePath();
        serviceConfiguration.setBrokerClientTlsTrustStorePassword(MockedPulsarServiceBaseTest.BROKER_KEYSTORE_PW);
    }