Client switch connection between different pulsar cluster.

[codelipenghui]

It's necessary to add a mechanism to support client switch connection between different pulsar cluster.

Motivation

Generally, we can use DNS or TCP proxy to improve broker availability when you have many pulsar cluster and one cluster is not available(such as network error). But in this solution, we need a backup cluster, the backup cluster has the same configurations. And in this solution have a problem with only can close connection by broker(shut down the broker or setting the network or offload namespace) to trigger client reconnect. So i have a idea, we can trigger client reconnect by pulsar client.

Implement

By now, we can create a pulsar client use service url. We can add a method named serviceUrlProvider to create a pulsar client.

public interface ServiceUrlProvider {

    /**
     * Get pulsar service url from ServiceUrlProvider.
     *
     * @return pulsar service url.
     */
    String getServiceUrl();

    /**
     * Set pulsar client to the provider for provider can control the pulsar client,
     * such as {@link PulsarClient#forceCloseConnection()} or {@link PulsarClient#close()}.
     *
     * @param client created pulsar client.
     */
    void setClient(PulsarClient client);

}

We can create custom provider by ServiceUrlProvider such as ZookeeperServiceUrlProvider. ZookeeperServiceUrlProvider can watch zookeeper node update event then trigger pulsar client to reconnect the new service url. We can add method like this:

public void onServiceUrlChanged(String newServiceUrl) throws PulsarClientException {
            this.getPulsarClient().getConf().setServiceUrl(newServiceUrl);
            this.getPulsarClient().reloadLookUp();
            this.getPulsarClient().forceCloseConnection();
        }

In this solution we switch client connection simpler and no need to do any thing in pulsar broker.

Already have a PR for this issue 2543

[sijie]

This issue is being implemented on #2543

[Carl-Hee]

@sijie Is dns necessary, or can we use IP address directly?

[codelipenghui]

#2671 Refactored to simplify API for adding ServiceUrlProvider

use

this.getPulsarClient().updateServiceUrl(newServiceUrl);

replace

this.getPulsarClient().getConf().setServiceUrl(newServiceUrl);
this.getPulsarClient().reloadLookUp();
this.getPulsarClient().forceCloseConnection();

[sijie]

This is fixed by #2543 #2671