This is the voting thread for PIP-121. It will stay open for at least 48 hours.
https://github.com/apache/pulsar/issues/13315 Pasted below for quoting convenience. ----- ### Motivation We have geo-replication to support Pulsar cluster level failover. We can set up Pulsar cluster A as a primary cluster in data center A, and setup Pulsar cluster B as backup cluster in data center B. Then we configure geo-replication between cluster A and cluster B. All the clients are connected to the Pulsar cluster by DNS. If cluster A is down, we should switch the DNS to point the target Pulsar cluster from cluster A to cluster B. After the clients are resolved to cluster B, they can produce and consume messages normally. After cluster A recovers, the administrator should switch the DNS back to cluster A. However, the current method has two shortcomings. 1. The administrator should monitor the status of all Pulsar clusters, and switch the DNS as soon as possible when cluster A is down. The switch and recovery is not automatic and recovery time is controlled by the administrator, which will put the administrator under heavy load. 2. The Pulsar client and DNS system have a cache. When the administrator switches the DNS from cluster A to Cluster B, it will take some time for cache trigger timeout, which will delay client recovery time and lead to the product/consumer message failing. ### Goal It's better to provide an automatic cluster level failure recovery mechanism to make pulsar cluster failover more effective. We should support pulsar clients auto switching from cluster A to cluster B when it detects cluster A has been down according to the configured detecting policy and switch back to cluster A when it has recovered. The reason why we should switch back to cluster A is that most applications may be deployed in data center A and they have low network cost for communicating with pulsar cluster A. If they keep visiting pulsar cluster B, they have high network cost, and cause high produce/consume latency. In order to improve the DNS cache problem, we should provide an administrator controlled switch provider for administrators to update service URLs. In the end, we should provide an auto service URL switch provider and administrator controlled switch provider. ### Design We have already provided the `ServiceUrlProvider` interface to support different service URLs. In order to support automatic cluster level failure auto recovery, we can provide different ServiceUrlProvider implementations. For current requirements, we can provide `AutoClusterFailover` and `ControlledClusterFailover`. #### AutoClusterFailover In order to support auto switching from the primary cluster to the secondary, we can provide a probe task, which will probe the activity of the primary cluster and the secondary one. When it finds the primary cluster failed more than `failoverDelayMs`, it will switch to the secondary cluster by calling `updateServiceUrl`. After switching to the secondary cluster, the `AutoClusterFailover` will continue to probe the primary cluster. If the primary cluster comes back and remains active for `switchBackDelayMs`, it will switch back to the primary cluster. The APIs are listed as follows. In order to support multiple secondary clusters, use List to store secondary cluster urls. When the primary cluster probe fails for failoverDelayMs, it will start to probe the secondary cluster list one by one, once it finds the active cluster, it will switch to the target cluster. Notice: If you configured multiple clusters, you should turn on cluster level geo-replication to ensure the topic data sync between all primary and secondary clusters. Otherwise, it may distribute the topic data into different clusters. And the consumers won’t get the whole data of the topic. In order to support different authentication configurations between clusters, we provide the authentication relation configurations updated with the target cluster. ```Java public class AutoClusterFailover implements ServiceUrlProvider { private AutoClusterFailover(AutoClusterFailoverBuilderImpl builder) { // } @Override public void initialize(PulsarClient client) { this.pulsarClient = client; // start to probe primary cluster active or not executor.scheduleAtFixedRate(catchingAndLoggingThrowables(() -> { // probe and switch }), intervalMs, intervalMs, TimeUnit.MILLISECONDS); } @Override public String getServiceUrl() { return this.currentPulsarServiceUrl; } @Override public void close() { this.executor.shutdown(); } // probe pulsar cluster available private boolean probeAvailable(String url, int timeout) { } ``` In order to create an `AutoClusterFailover` instance, we use `AutoClusterFailoverBuilder` interface to build the target instance. The `AutoClusterFailoverBuilder` interface is located in the `pulsar-client-api` package. In the `probeAvailable` method, we will probe the Pulsar service port, and check whether the port is open. This probe method has many disadvantages, such as We're connecting to a Pulsar proxy, but there are no available brokers Using Istio on the server side, which always accepts the connection even if the broker is in a bad state We might have deadlocks in (all) brokers and while the connections get accepted, the brokers are not able to serve them. In order to solve this problem, we’d better provide a health check command on the broker side, just like Zookeeper’s `ruok` command. We can use the probe port method first, and in the next step, we will provide the health check command on the broker side. #### ControlledClusterFailover If the users want to control the cluster switch operation, they can provide the current service URL by a http service. The `ControlledClusterFailover` will get the newest service url from the provided http service periodically. The APIs are listed as follows. ```Java public class ControlledClusterFailover implements ServiceUrlProvider { private ControlledClusterFailover(String defaultServiceUrl, String urlProvider) throws IOException { } @Override public void initialize(PulsarClient client) { this.pulsarClient = client; // start to check service url every 30 seconds executor.scheduleAtFixedRate(catchingAndLoggingThrowables(() -> { // probe and switch }), interval, interval, TimeUnit.MILLISECONDS); } protected ControlledConfiguration fetchControlledConfiguration() throws IOException { // call the service to get controlled configuration } @Override public String getServiceUrl() { return this.currentPulsarServiceUrl; } @Override public void close() { this.executor.shutdown(); } protected static class ControlledConfiguration { private String serviceUrl; private String tlsTrustCertsFilePath; private String authPluginClassName; private String authParamsString; } ``` The configuration we get from the third url provider, we define it as java Bean by json format. In the configuration, we provide authentication-related parameters to support different clusters that have different authentication configurations. These authentication-related parameters can support all current authentication plugin types. In order to create an `ControlledClusterFailover` instance, we use the `ControlledClusterFailoverBuilder` interface to build the target instance. The `ControlledClusterFailoverBuilder` interface is located in the `pulsar-client-api` package. ### API Changes For the current `ServiceUrlProvider` interface, we should add a `close` method to close an allocated resource, such as a timer thread. ```Java public interface ServiceUrlProvider { /** * Close the resource that the provider allocated. * */ default void close() { // do nothing } /** * Update the authentication this client is using. * * @param authentication * * @throws IOException */ void updateAuthentication(Authentication authentication) throws IOException; /** * Update the tlsTrustCertsFilePath this client is using. * * @param tlsTrustCertsFilePath */ void updateTlsTrustCertsFilePath(String tlsTrustCertsFilePath); /** * Update the tlsTrustStorePath and tlsTrustStorePassword this client is using. * * @param tlsTrustStorePath * @param tlsTrustStorePassword */ void updateTlsTrustStorePathAndPassword(String tlsTrustStorePath, String tlsTrustStorePassword); } ``` ### Tests Add tests for the two service provider implementations. For `AutoClusterFailover`, when the primary cluster shuts down, it should switch to the secondary cluster. And then the primary cluster came back, we should switch back. For `ControlledClusterFailover`, when switching the service url on the http service side, it should switch to the newest service url. ### Implementation Prototype implementation PR: https://github.com/apache/pulsar/pull/13316
