PIP-121: Pulsar cluster level auto failover on client side #13316

hangc0276 · 2021-12-15T02:23:06Z

Related to #13315

Modification

add Pulsar cluster level auto failover

github-actions · 2021-12-15T02:23:29Z

@hangc0276:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

github-actions · 2021-12-15T02:26:02Z

@hangc0276:Thanks for providing doc info!

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

fantapsody · 2021-12-15T02:53:30Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ServiceUrlProvider.java

+     * Close the resource that the provider allocated.
+     *
+     */
+    default void close() {


Would it be better to make the interface inherit AutoCloseable?

So, we can remove the default empty implementation here?

Other serviceUrlProvider implementation also need to implement close() method, so I add the default implementation.

fantapsody · 2021-12-15T02:56:50Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+        this.primaryFailedTimestamp = -1;
+        this.primaryRecoverTimestamp = -1;
+        this.secondaryFailedTimestamp = -1;
+        this.timer = new Timer("pulsar-service-provider");


Would it be better to use ScheduledThreadPoolExecutor here?

fantapsody · 2021-12-15T03:04:50Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+                    }
+
+                    if (primaryFailedTimestamp == -1) {
+                        primaryFailedTimestamp = System.currentTimeMillis();


It might be better to use System.nanoTime() to calculate the time locally to avoid time drifts.

fantapsody · 2021-12-15T03:45:17Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+            public void run() {
+                // current pulsar serviceUrl is primary
+                if (currentPulsarServiceUrl.equals(primary)) {
+                    if (probeAvailable(primary, timeout)) {


How about failover after a succession of failures (3 for example, and should be configurable) to avoid unnecessary switches?

The current implementation is switch by failure delay time, it can control the client failover time. If we use failure times, the failover time will out of control.

fantapsody · 2021-12-15T03:50:31Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java

+        try {
+            URLConnection conn = pulsarUrlProvider.openConnection();
+            inputStream = conn.getInputStream();
+            return new String(IOUtils.toByteArray(inputStream), StandardCharsets.UTF_8);


Using the service url as the response body might not be very extensible.

👍 We should use a JSON response here

fantapsody · 2021-12-15T03:55:31Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+            }
+
+            Socket socket = new Socket();
+            socket.connect(new InetSocketAddress(parseHost(hostAndPort), parsePort(hostAndPort)), timeout);


It might be better to make a request to the server to ensure it is actually in a good state.

The current implementation is probe whether the service port is open or not. If we want to probe the broker state, it's better to provide a health check command for broker service on broker side. I can use another PR to provide this feature on broker side. @merlimat Do you have any other ideas?

zymap

As you mentioned, we use geo-replication to support cluster failover, that means there might be have more than two cluster. In the implementation, we only can configure two clusters for the failover. I am thinking of, do we need to support more clusters to failover to make the cluster is more robust?

zymap · 2021-12-16T02:10:14Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+    private static String parseHostAndPort(String url) {
+        if (Strings.isNullOrEmpty(url) || !url.startsWith("pulsar")) {
+            throw new IllegalArgumentException("'" + url + "' isn't an Pulsar service URL");
+        }
+
+        int uriSeparatorPos = url.indexOf("://");
+        if (uriSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + url + "' isn't an URI.");
+        }
+        return url.substring(uriSeparatorPos + 3);
+    }
+
+    private static String parseHost(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return hostAndPort.substring(0, portSeparatorPos);
+    }
+
+    private static Integer parsePort(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return Integer.valueOf(hostAndPort.substring(portSeparatorPos+1));
+    }


Would it be better to use URL class to parse a URL and you can get the host and port easily.

I think URI could be better here, because URL assumes a "known" protocol scheme.

@hangc0276 please address this comment

zymap · 2021-12-16T02:32:41Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+                                currentPulsarServiceUrl, System.currentTimeMillis() - primaryFailedTimestamp);
+                    }
+                } else { // current pulsar service URL is secondary, probe whether we need to switch back to primary.
+                    if (!probeAvailable(currentPulsarServiceUrl, timeout)) {


Can we make the url check and switch as a method? I think most of the following logic is duplicated with the primary service url check?

hpvd · 2021-12-16T07:25:05Z

As you mentioned, we use geo-replication to support cluster failover, that means there might be have more than two cluster. In the implementation, we only can configure two clusters for the failover. I am thinking of, do we need to support more clusters to failover to make the cluster is more robust?

Maybe one could even go further and automatically choose from all of the replicated clusters, the nearest one (via latency probe/from pre-probed list)?
(as follow-up step of this feature)

hangc0276 · 2021-12-17T02:55:36Z

As you mentioned, we use geo-replication to support cluster failover, that means there might be have more than two cluster. In the implementation, we only can configure two clusters for the failover. I am thinking of, do we need to support more clusters to failover to make the cluster is more robust?

@zymap if we have multi secondary clusters, different clients will switch to different secondary clusters. There are some problems.

In geo-replication case, it depends on geo-replication to ensure data integrity. It can't ensure message order in one topic. If all the producer just produce to the same cluster, we can ensure message order in one topic.
If the user doesn't enable geo-replication, we can't ensure the consumer get the whole messages from the topic. Because the producer may switched to cluster B, and the consumer may switched to cluster C.

So, in current implementation, we can support only one secondary cluster first.

hangc0276 · 2021-12-17T02:56:33Z

As you mentioned, we use geo-replication to support cluster failover, that means there might be have more than two cluster. In the implementation, we only can configure two clusters for the failover. I am thinking of, do we need to support more clusters to failover to make the cluster is more robust?

Maybe one could even go further and automatically choose from all of the replicated clusters, the nearest one (via latency probe/from pre-probed list)? (as follow-up step of this feature)

@hpvd Please refer to #13316 (comment)

merlimat · 2021-12-21T15:19:58Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AutoClusterFailoverBuilder.java

+     * @param secondary
+     * @return
+     */
+    AutoClusterFailoverBuilder secondary(String secondary);


We could take a list/set of serviceUrls, just to make it more general

We could take a list/set of serviceUrls, just to make it more general

done

merlimat · 2021-12-21T17:17:10Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+            }
+
+            Socket socket = new Socket();
+            socket.connect(new InetSocketAddress(parseHost(hostAndPort), parsePort(hostAndPort)), TIMEOUT);


This will check for TCP connectivity, which is a good start, though there can be many cases in which it will give the false impression that cluster is healthy:

We're connecting to a Pulsar proxy, but there are no available brokers

Using Istio on server side, which always accept 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.

We should consider to have a more in depth test to:

Check that we can authenticate with brokers

(Maybe) Estimate how many brokers are up & healthy in the cluster

Yes, the current probe port method has many disadvantages. We can use this method first, and in next step, we'd better provide a health check command, just like Zookeeper ruok command, to probe the healthy of the target Pulsar cluster.

merlimat · 2021-12-21T17:17:47Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+    private static String parseHostAndPort(String url) {
+        if (Strings.isNullOrEmpty(url) || !url.startsWith("pulsar")) {
+            throw new IllegalArgumentException("'" + url + "' isn't an Pulsar service URL");
+        }
+
+        int uriSeparatorPos = url.indexOf("://");
+        if (uriSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + url + "' isn't an URI.");
+        }
+        return url.substring(uriSeparatorPos + 3);
+    }
+
+    private static String parseHost(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return hostAndPort.substring(0, portSeparatorPos);
+    }
+
+    private static Integer parsePort(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return Integer.valueOf(hostAndPort.substring(portSeparatorPos+1));
+    }


I think URI could be better here, because URL assumes a "known" protocol scheme.

merlimat · 2021-12-21T17:18:33Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java

+        try {
+            URLConnection conn = pulsarUrlProvider.openConnection();
+            inputStream = conn.getInputStream();
+            return new String(IOUtils.toByteArray(inputStream), StandardCharsets.UTF_8);


👍 We should use a JSON response here

merlimat · 2021-12-21T17:20:05Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java

+    private volatile String currentPulsarServiceUrl;
+    private final URL pulsarUrlProvider;
+    private final ScheduledExecutorService executor;
+    private final int interval = 30_000;


We should take the poll interval as a config in the builder

We should take the poll interval as a config in the builder

done

zymap · 2022-01-05T01:20:34Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+    private final long intervalMs;
+    private static final int TIMEOUT = 30_000;
+
+    private AutoClusterFailover(String primary, List<String> secondary, long failoverDelayNs, long switchBackDelayNs,


Make the arguments as configuration data or something else to avoid passing so many configurations into the constructor?

zymap · 2022-01-05T01:26:37Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AutoClusterFailoverBuilder.java

+     * @param authentication
+     * @return
+     */
+    AutoClusterFailoverBuilder secondaryAuthentication(List<Authentication> authentication);


Is it necessary to use a map to keep the secondary URLs and the authentication? I think we should consider users will not add the authentication in order with the secondary URL lists.

zymap · 2022-01-05T01:26:45Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AutoClusterFailoverBuilder.java

+     * @param tlsTrustCertsFilePath
+     * @return
+     */
+    AutoClusterFailoverBuilder secondaryTlsTrustCertsFilePath(List<String> tlsTrustCertsFilePath);


same above.

zymap · 2022-01-05T01:26:55Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AutoClusterFailoverBuilder.java

+     * @param tlsTrustStorePath
+     * @return
+     */
+    AutoClusterFailoverBuilder secondaryTlsTrustStorePath(List<String> tlsTrustStorePath);


same above.

zymap · 2022-01-05T01:27:06Z

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AutoClusterFailoverBuilder.java

+     * @param tlsTrustStorePassword
+     * @return
+     */
+    AutoClusterFailoverBuilder secondaryTlsTrustStorePassword(List<String> tlsTrustStorePassword);


zymap · 2022-01-05T02:25:41Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java

+    private static String parseHostAndPort(String url) {
+        if (Strings.isNullOrEmpty(url) || !url.startsWith("pulsar")) {
+            throw new IllegalArgumentException("'" + url + "' isn't an Pulsar service URL");
+        }
+
+        int uriSeparatorPos = url.indexOf("://");
+        if (uriSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + url + "' isn't an URI.");
+        }
+        return url.substring(uriSeparatorPos + 3);
+    }
+
+    private static String parseHost(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return hostAndPort.substring(0, portSeparatorPos);
+    }
+
+    private static Integer parsePort(String hostAndPort) {
+        int portSeparatorPos = hostAndPort.indexOf(":");
+        if (portSeparatorPos == -1) {
+            throw new IllegalArgumentException("'" + hostAndPort + "' isn't an URI.");
+        }
+        return Integer.valueOf(hostAndPort.substring(portSeparatorPos+1));
+    }


@hangc0276 please address this comment

zymap · 2022-01-05T02:37:11Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -746,7 +754,7 @@ public void shutdown() throws PulsarClientException {
            } catch (PulsarClientException e) {
                throwable = e;
            }
-            if (conf != null && conf.getAuthentication() != null) {
+            if (conf.getAuthentication() != null) {


Why remove the conf != null ?

I add it back.

zymap · 2022-01-05T02:51:55Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ControlledClusterFailover.java

+                        String authParamsString = controlledConfiguration.getAuthParamsString();
+                        String token = controlledConfiguration.getToken();
+
+                        switch (authPluginClassName) {


I remembered that we can configure the authPlugin and authParams for all auth plugins, why do we need to case the auth plugins? Does the AuthenticationFactory.create(authPlugin, authParam) can not handle all things?

hangc0276 · 2022-01-06T14:49:36Z

As you mentioned, we use geo-replication to support cluster failover, that means there might be have more than two cluster. In the implementation, we only can configure two clusters for the failover. I am thinking of, do we need to support more clusters to failover to make the cluster is more robust?

Maybe one could even go further and automatically choose from all of the replicated clusters, the nearest one (via latency probe/from pre-probed list)? (as follow-up step of this feature)

@hpvd I added the SecondaryChoosePolicy interface to support different choose policy. Currently we support ORDER first, and we can support more choose policy in the following PRs.

… dependency

) Related to apache#13315 ### Modification 1. add Pulsar cluster level auto failover

github-actions bot assigned hangc0276 Dec 15, 2021

github-actions bot added the doc-label-missing label Dec 15, 2021

hangc0276 changed the title ~~[PIP-121] Pulsar cluster level auto failover [WIP]~~ PIP-121: Pulsar cluster level auto failover [WIP] Dec 15, 2021

hangc0276 added the doc-required Your PR changes impact docs and you will update later. label Dec 15, 2021

github-actions bot removed the doc-label-missing label Dec 15, 2021

merlimat reviewed Dec 15, 2021

View reviewed changes

pulsar-client/src/main/java/org/apache/pulsar/client/impl/AutoClusterFailover.java Show resolved Hide resolved

merlimat added the release/note-required label Dec 15, 2021

merlimat modified the milestones: 2.11.0, 2.10.0 Dec 15, 2021

fantapsody reviewed Dec 15, 2021

View reviewed changes

zymap reviewed Dec 16, 2021

View reviewed changes

hangc0276 changed the title ~~PIP-121: Pulsar cluster level auto failover [WIP]~~ PIP-121: Pulsar cluster level auto failover Dec 16, 2021

merlimat reviewed Dec 21, 2021

View reviewed changes

hangc0276 requested review from merlimat and zymap January 4, 2022 09:12

zymap reviewed Jan 5, 2022

View reviewed changes

hangc0276 requested review from congbobo184, 315157973, codelipenghui, eolivelli, gaoran10 and zymap January 5, 2022 07:28

hpvd mentioned this pull request Jan 5, 2022

PIP-121: Pulsar cluster level auto failover on client side #13315

Closed

hangc0276 added 17 commits January 29, 2022 13:30

fix a bug

a8f75d5

fix a bug

37fd325

simplify code and add Builder interface in pulsar-client-api

f362c00

add tests

5ef9265

add tests to cover service url switch

f1f569e

tune code

04abd2b

add authentication support

179372f

add authentication support

fa176bb

support multi secondary cluster

ee812ff

format code

abfe14f

address comments

fecb2df

address comments

8017d67

add secondaryChoosePolicy support

1c8730f

add default value doc

1f7f9a8

support http header

5f50a28

format code

f5f162b

use AsyncHttpClient to perform http request, avoiding introduce other…

5d9b48b

… dependency

hangc0276 force-pushed the chenhang/service_url_provider branch from 967ace6 to 5d9b48b Compare January 29, 2022 05:31

merlimat approved these changes Jan 29, 2022

View reviewed changes

codelipenghui merged commit 6e353b7 into apache:master Jan 29, 2022

Anonymitaet mentioned this pull request Feb 18, 2022

[Doc] add doc for cluster-level failover #14314

Merged

Anonymitaet added doc-complete Your PR changes impact docs and the related docs have been already added. and removed doc-required Your PR changes impact docs and you will update later. labels Feb 18, 2022

Nicklee007 pushed a commit to Nicklee007/pulsar that referenced this pull request Apr 20, 2022

PIP-121: Pulsar cluster level auto failover on client side (apache#13316

1260a89

) Related to apache#13315 ### Modification 1. add Pulsar cluster level auto failover

coderzc mentioned this pull request May 31, 2022

PIP-172 : Introduce the HEALTH_CHECK command in the binary protocol #15859

Open

sijie mentioned this pull request May 31, 2022

ISSUE-15859: PIP-172 : Introduce the HEALTH_CHECK command in the binary protocol streamnative/pulsar-archived#4307

Open

This was referenced Dec 30, 2022

[feature request] Pulsar cluster level auto failover on client side apache/pulsar-client-cpp#164

Open

[feature request] Pulsar cluster level auto failover on client side apache/pulsar-client-go#929

Open

LeoSht mentioned this pull request Jan 17, 2023

Add support for Pulsar cluster level auto failover (PIP-121) fsprojects/pulsar-client-dotnet#217

Open

lhotari mentioned this pull request Feb 29, 2024

[improve][broker]PIP-340 Optimization of Probe Implementation for Automatic Failover #22133

Open

15 tasks

PIP-121: Pulsar cluster level auto failover on client side #13316

PIP-121: Pulsar cluster level auto failover on client side #13316

Conversation

hangc0276 commented Dec 15, 2021

Modification

github-actions bot commented Dec 15, 2021

github-actions bot commented Dec 15, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zymap left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hpvd commented Dec 16, 2021 • edited Loading

hangc0276 commented Dec 17, 2021

hangc0276 commented Dec 17, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hangc0276 commented Jan 6, 2022

hpvd commented Dec 16, 2021 •

edited

Loading