Skip to content

[fix][broker] Ignore non-retryable errors when loading topic-level policies before loading topics #22014

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
BewareMyPower wants to merge 1 commit into from
Closed

[fix][broker] Ignore non-retryable errors when loading topic-level policies before loading topics #22014

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -970,6 +971,11 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo
final CompletableFuture<Optional<TopicPolicies>> 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());
Expand Down
22 changes: 18 additions & 4 deletions ...er/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
Expand All @@ -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) {
Expand Down Expand Up @@ -93,10 +104,13 @@ protected void setup() throws Exception {
Set<String> 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();
Expand Down
1 change: 1 addition & 0 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down