@Override
    public void initialize(PulsarClient client) {
        this.pulsarClient = (PulsarClientImpl) client;
        ClientConfigurationData config = pulsarClient.getConfiguration();
        if (config != null) {
            this.primaryAuthentication = config.getAuthentication();
            this.primaryTlsTrustCertsFilePath = config.getTlsTrustCertsFilePath();
            this.primaryTlsTrustStorePath = config.getTlsTrustStorePath();
            this.primaryTlsTrustStorePassword = config.getTlsTrustStorePassword();
        }

        // start to probe primary cluster active or not
        this.executor.scheduleAtFixedRate(catchingAndLoggingThrowables(() -> {
            if (currentPulsarServiceUrl.equals(primary)) {
                // current service url is primary, probe whether it is down
                probeAndUpdateServiceUrl(secondary, secondaryAuthentications, secondaryTlsTrustCertsFilePaths,
                        secondaryTlsTrustStorePaths, secondaryTlsTrustStorePasswords);
            } else {
                // current service url is secondary, probe whether it is down
                probeAndUpdateServiceUrl(primary, primaryAuthentication, primaryTlsTrustCertsFilePath,
                        primaryTlsTrustStorePath, primaryTlsTrustStorePassword);
                // secondary cluster is up, check whether need to switch back to primary or not
                if (!currentPulsarServiceUrl.equals(primary)) {
                    probeAndCheckSwitchBack(primary, primaryAuthentication, primaryTlsTrustCertsFilePath,
                            primaryTlsTrustStorePath, primaryTlsTrustStorePassword);
                }
            }
        }), intervalMs, intervalMs, TimeUnit.MILLISECONDS);

    }
@Test
    public void testInitialize() throws Exception {
        String primary = "pulsar://localhost:6650";
        String secondary = "pulsar://localhost:6651";
        long failoverDelay = 10;
        long switchBackDelay = 10;
        long checkInterval = 1_000;

        ClientConfigurationData configurationData = new ClientConfigurationData();

        @Cleanup
        ServiceUrlProvider provider = AutoClusterFailover.builder()
                .primary(primary)
                .secondary(Collections.singletonList(secondary))
                .failoverDelay(failoverDelay, TimeUnit.MILLISECONDS)
                .switchBackDelay(switchBackDelay, TimeUnit.MILLISECONDS)
                .checkInterval(checkInterval, TimeUnit.MILLISECONDS)
                .build();

        AutoClusterFailover autoClusterFailover = Mockito.spy((AutoClusterFailover) provider);
        PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class);
        ConnectionPool connectionPool = mock(ConnectionPool.class);
        when(pulsarClient.getCnxPool()).thenReturn(connectionPool);
        Mockito.doReturn(false).when(autoClusterFailover).probeAvailable(primary);
        Mockito.doReturn(true).when(autoClusterFailover).probeAvailable(secondary);
        Mockito.doReturn(configurationData).when(pulsarClient).getConfiguration();

        autoClusterFailover.initialize(pulsarClient);

        for (int i = 0; i < 2; i++) {
            Awaitility.await().untilAsserted(() ->
                    assertEquals(autoClusterFailover.getServiceUrl(), secondary));
            assertEquals(autoClusterFailover.getFailedTimestamp(), -1);

            // primary cluster came back
            Mockito.doReturn(true).when(autoClusterFailover).probeAvailable(primary);
            Awaitility.await().untilAsserted(() ->
                    assertEquals(autoClusterFailover.getServiceUrl(), primary));
            assertEquals(autoClusterFailover.getRecoverTimestamp(), -1);
            assertEquals(autoClusterFailover.getFailedTimestamp(), -1);

            Mockito.doReturn(false).when(autoClusterFailover).probeAvailable(primary);
        }
    }