From a548a14114d9ab4fef8d83c9f3e4b365d9a5ff76 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Fri, 2 Feb 2024 20:57:11 +0800 Subject: [PATCH] [fix][broker] Ignore non-retryable errors when loading topic-level policies before loading topics ### Motivation https://github.com/apache/pulsar/pull/21445 introduces an implicit requirement that if the topic policies cannot be loaded, `BrokerService#getTopic` will fail and then the client will retry loading the topic. It could break some existing usages like the tests in the C++ client: https://github.com/apache/pulsar-client-cpp/pull/394 This change is applied only to avoid the race condition when unloading a namespace bundle. However, the client should not fail if the topic-level policies are not available. ### Modifications If the topic policies cannot be loaded due to a non-retryable error, we should not fail the `getTopic`. For retryable errors, the client will still retry until the broker gets the topic-level policies successfully after some attempts. Modify `TokenAuthenticatedProducerConsumerTest` to protect the change. --- .../pulsar/broker/service/BrokerService.java | 6 +++++ ...okenAuthenticatedProducerConsumerTest.java | 22 +++++++++++++++---- .../client/api/PulsarClientException.java | 1 + 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index 0383c63b1f3bc..d77e224157a77 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -133,6 +133,7 @@ import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.api.ClientBuilder; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.SizeUnit; import org.apache.pulsar.client.impl.ClientBuilderImpl; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; @@ -970,6 +971,11 @@ public CompletableFuture> getTopic(final TopicName topicName, bo final CompletableFuture> topicPoliciesFuture = getTopicPoliciesBypassSystemTopic(topicName); return topicPoliciesFuture.exceptionally(ex -> { + if (!PulsarClientException.isRetriableError(ex.getCause())) { + // Topic policies are not available, we should not prevent the creation of this topic + log.warn("Failed to get topic policies due to non-retriable error: {}", ex.getMessage()); + return Optional.empty(); + } final Throwable rc = FutureUtil.unwrapCompletionException(ex); final String errorInfo = String.format("Topic creation encountered an exception by initialize" + " topic policies service. topic_name=%s error_message=%s", topicName, rc.getMessage()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java index f8ae0279e08b7..d593f8d1d4b3f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java @@ -45,6 +45,7 @@ import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; +import org.testng.annotations.Factory; import org.testng.annotations.Test; /** @@ -58,14 +59,24 @@ public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase private final String ADMIN_TOKEN; private final String TOKEN_PUBLIC_KEY; + private final boolean brokerClientAuthEnabled; + + @Factory + public static Object[] instances() throws NoSuchAlgorithmException { + return new Object[] { + new TokenAuthenticatedProducerConsumerTest(false), + new TokenAuthenticatedProducerConsumerTest(true), + }; + } - TokenAuthenticatedProducerConsumerTest() throws NoSuchAlgorithmException { + TokenAuthenticatedProducerConsumerTest(boolean brokerClientAuthEnabled) throws NoSuchAlgorithmException { KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA"); KeyPair kp = kpg.generateKeyPair(); byte[] encodedPublicKey = kp.getPublic().getEncoded(); TOKEN_PUBLIC_KEY = "data:;base64," + Base64.getEncoder().encodeToString(encodedPublicKey); ADMIN_TOKEN = generateToken(kp); + this.brokerClientAuthEnabled = brokerClientAuthEnabled; } private String generateToken(KeyPair kp) { @@ -93,10 +104,13 @@ protected void setup() throws Exception { Set providers = new HashSet<>(); providers.add(AuthenticationProviderToken.class.getName()); conf.setAuthenticationProviders(providers); - conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName()); - conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN); + if (brokerClientAuthEnabled) { + conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName()); + conf.setBrokerClientAuthenticationParameters("token:" + ADMIN_TOKEN); + } - conf.setClusterName("test"); + final var clusterName = "test"; + conf.setClusterName(clusterName); // Set provider domain name Properties properties = new Properties(); diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java index 007308ec7ab46..369e4ecbcef01 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java @@ -1158,6 +1158,7 @@ public void setSequenceId(long sequenceId) { public static boolean isRetriableError(Throwable t) { if (t instanceof AuthorizationException + || t instanceof AuthenticationException || t instanceof InvalidServiceURL || t instanceof InvalidConfigurationException || t instanceof NotFoundException