Skip to content

[Issue 10300][broker] Fix authorization error if partition number of partitioned topic is updated. #10333

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

Merged
codelipenghui merged 1 commit into from
May 6, 2021
Merged

[Issue 10300][broker] Fix authorization error if partition number of partitioned topic is updated. #10333

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
21 changes: 19 additions & 2 deletions ...mon/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -434,16 +434,33 @@ public CompletableFuture<Boolean> checkPermission(TopicName topicName, String ro
return;
}
}

// If the partition number of the partitioned topic having topic level policy is updated,
// the new sub partitions may not inherit the policy of the partition topic.
// We can also check the permission of partitioned topic.
// For https://github.com/apache/pulsar/issues/10300
if (topicName.isPartitioned()) {
topicRoles = policies.get().auth_policies.destination_auth.get(topicName.getPartitionedTopicName());
if (topicRoles != null) {
// Topic has custom policy
Set<AuthAction> topicActions = topicRoles.get(role);
if (topicActions != null && topicActions.contains(action)) {
// The role has topic level permission
permissionFuture.complete(true);
return;
}
}
}
}
permissionFuture.complete(false);
}).exceptionally(ex -> {
log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
ex.getMessage());
permissionFuture.completeExceptionally(ex);
return null;
});
} catch (Exception e) {
log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
log.warn("Client with Role - {} failed to get permissions for topic - {}. {}", role, topicName,
e.getMessage());
permissionFuture.completeExceptionally(e);
}
Expand Down
91 changes: 91 additions & 0 deletions pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,97 @@ public void testProxyAuthorization() throws Exception {
log.info("-- Exiting {} test --", methodName);
}

/**
* <pre>
* 1. Create a 2-partition topic and grant produce/consume permission to client role.
* 2. Use producer/consumer with client role to process the topic, which is fine.
* 2. Update the topic partition number to 4.
* 3. Use new producer/consumer with client role to process the topic.
* 4. Broker should authorize producer/consumer normally.
* </pre>
*/
@Test
public void testUpdatePartitionNumAndReconnect() throws Exception {
log.info("-- Starting {} test --", methodName);

startProxy();
createAdminClient();
PulsarClient proxyClient = createPulsarClient(proxyService.getServiceUrl(), PulsarClient.builder());

String clusterName = "proxy-authorization";
String namespaceName = "my-property/my-ns";
String topicName = "persistent://my-property/my-ns/my-topic1";
String subscriptionName = "my-subscriber-name";

admin.clusters().createCluster(clusterName, new ClusterData(brokerUrl.toString()));

admin.tenants().createTenant("my-property",
new TenantInfo(Sets.newHashSet(), Sets.newHashSet(clusterName)));
admin.namespaces().createNamespace(namespaceName);
admin.topics().createPartitionedTopic(topicName, 2);
admin.topics().grantPermission(topicName, CLIENT_ROLE,
Sets.newHashSet(AuthAction.consume, AuthAction.produce));

Consumer<byte[]> consumer = proxyClient.newConsumer()
.topic(topicName)
.subscriptionName(subscriptionName).subscribe();

Producer<byte[]> producer = proxyClient.newProducer(Schema.BYTES)
.topic(topicName).create();
final int MSG_NUM = 10;
Set<String> messageSet = Sets.newHashSet();
for (int i = 0; i < MSG_NUM; i++) {
String message = "my-message-" + i;
messageSet.add(message);
producer.send(message.getBytes());
}

Message<byte[]> msg;
Set<String> receivedMessageSet = Sets.newHashSet();
for (int i = 0; i < MSG_NUM; i++) {
msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedMessage = new String(msg.getData());
log.debug("Received message: [{}]", receivedMessage);
String expectedMessage = "my-message-" + i;
receivedMessageSet.add(expectedMessage);
consumer.acknowledgeAsync(msg);
}
Assert.assertEquals(messageSet, receivedMessageSet);
consumer.close();
producer.close();

// update partition num
admin.topics().updatePartitionedTopic(topicName, 4);

// produce/consume the topic again
consumer = proxyClient.newConsumer()
.topic(topicName)
.subscriptionName(subscriptionName).subscribe();
producer = proxyClient.newProducer(Schema.BYTES)
.topic(topicName).create();

messageSet.clear();
for (int i = 0; i < MSG_NUM; i++) {
String message = "my-message-" + i;
messageSet.add(message);
producer.send(message.getBytes());
}

receivedMessageSet.clear();
for (int i = 0; i < MSG_NUM; i++) {
msg = consumer.receive(5, TimeUnit.SECONDS);
String receivedMessage = new String(msg.getData());
log.debug("Received message: [{}]", receivedMessage);
String expectedMessage = "my-message-" + i;
receivedMessageSet.add(expectedMessage);
consumer.acknowledgeAsync(msg);
}
Assert.assertEquals(messageSet, receivedMessageSet);
consumer.close();
producer.close();
log.info("-- Exiting {} test --", methodName);
}

/**
* <pre>
* It verifies jwt + Authentication + Authorization (client -> proxy -> broker).
Expand Down