You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/01 18:11:16 UTC
(pulsar) 26/38: [fix][broker] Align partitioned topic truncate checks with non-partitioned topics (#26776)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.2-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 198074b6de3ba083fd66ba453884fa8c8dfeedf4
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 30 09:58:56 2026 +0300
[fix][broker] Align partitioned topic truncate checks with non-partitioned topics (#26776)
(cherry picked from commit b8b57698a33835c2394409e93bf7d704b779f914)
---
.../broker/admin/impl/PersistentTopicsBase.java | 28 +++++-----
.../apache/pulsar/broker/admin/TopicAuthZTest.java | 62 +++++++++++++++++++++-
2 files changed, 75 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index f1c3056f2bc8..2e8f33bffd50 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -4883,19 +4883,13 @@ public class PersistentTopicsBase extends AdminResource {
resumeAsyncResponseExceptionally(asyncResponse, cause);
}
- protected CompletableFuture<Void> internalTruncateNonPartitionedTopicAsync(boolean authoritative) {
- return validateAdminAccessForTenantAsync(topicName.getTenant())
- .thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
- .thenCompose(__ -> getTopicReferenceAsync(topicName))
- .thenCompose(Topic::truncate);
- }
-
protected CompletableFuture<Void> internalTruncateTopicAsync(boolean authoritative) {
-
- // If the topic name is a partition name, no need to get partition topic metadata again
- if (topicName.isPartitioned()) {
- return internalTruncateNonPartitionedTopicAsync(authoritative);
- } else {
+ // Validate tenant admin access once for partitioned, non-partitioned and partition topics
+ return validateAdminAccessForTenantAsync(topicName.getTenant()).thenCompose(__ -> {
+ // If the topic name is a partition name, no need to get partition topic metadata again
+ if (topicName.isPartitioned()) {
+ return truncateNonPartitionedTopicWithoutAccessCheckAsync(authoritative);
+ }
return getPartitionedTopicMetadataAsync(topicName, authoritative, false).thenCompose(meta -> {
if (meta.partitions > 0) {
final List<CompletableFuture<Void>> futures = new ArrayList<>(meta.partitions);
@@ -4912,10 +4906,16 @@ public class PersistentTopicsBase extends AdminResource {
}
return FutureUtil.waitForAll(futures);
} else {
- return internalTruncateNonPartitionedTopicAsync(authoritative);
+ return truncateNonPartitionedTopicWithoutAccessCheckAsync(authoritative);
}
});
- }
+ });
+ }
+
+ private CompletableFuture<Void> truncateNonPartitionedTopicWithoutAccessCheckAsync(boolean authoritative) {
+ return validateTopicOwnershipAsync(topicName, authoritative)
+ .thenCompose(__ -> getTopicReferenceAsync(topicName))
+ .thenCompose(Topic::truncate);
}
protected void internalSetReplicatedSubscriptionStatus(AsyncResponse asyncResponse, String subName,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
index fce47ba8311d..70556cc781e3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java
@@ -149,7 +149,7 @@ public class TopicAuthZTest extends AuthZTest {
public void testUnloadAndCompactAndTrim() {
final String random = UUID.randomUUID().toString();
final String topic = "persistent://public/default/" + random;
- final String subject = UUID.randomUUID().toString();
+ final String subject = UUID.randomUUID().toString();
final String token = Jwts.builder()
.claim("sub", subject).signWith(SECRET_KEY).compact();
superUserAdmin.topics().createPartitionedTopic(topic, 2);
@@ -2308,6 +2308,66 @@ public class TopicAuthZTest extends AuthZTest {
deleteTopic(topic, false);
}
+ @Test(dataProvider = "partitioned")
+ @SneakyThrows
+ public void testTruncate(boolean partitioned) {
+ final String random = UUID.randomUUID().toString();
+ final String topic = "persistent://public/default/" + random;
+ final String subject = UUID.randomUUID().toString();
+ final String token = Jwts.builder()
+ .claim("sub", subject).signWith(SECRET_KEY).compact();
+ final String subName = "test-sub";
+ createTopic(topic, partitioned);
+ superUserAdmin.topics().createSubscription(topic, subName, MessageId.earliest);
+ @Cleanup
+ final PulsarAdmin subAdmin = PulsarAdmin.builder()
+ .serviceHttpUrl(getPulsarService().getWebServiceAddress())
+ .authentication(new AuthenticationToken(token))
+ .build();
+ @Cleanup
+ final PulsarClient pulsarClient = PulsarClient.builder()
+ .serviceUrl(getPulsarService().getBrokerServiceUrl())
+ .authentication(new AuthenticationToken(SUPER_USER_TOKEN))
+ .build();
+ final int numMessages = 4;
+ try (Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic).enableBatching(false).create()) {
+ for (int i = 0; i < numMessages; i++) {
+ producer.send("msg-" + i);
+ }
+ }
+ Assert.assertEquals(getBacklog(topic, subName, partitioned), numMessages);
+
+ // Only super user and tenant admin can truncate, topic level permissions are not sufficient
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
+ () -> subAdmin.topics().truncate(topic));
+ for (AuthAction action : AuthAction.values()) {
+ superUserAdmin.topics().grantPermission(topic, subject, Set.of(action));
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
+ () -> subAdmin.topics().truncate(topic));
+ superUserAdmin.topics().revokePermissions(topic, subject);
+ }
+ superUserAdmin.topics().grantPermission(topic, subject, Set.of(AuthAction.produce, AuthAction.consume));
+ Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
+ () -> subAdmin.topics().truncate(topic));
+ superUserAdmin.topics().revokePermissions(topic, subject);
+ Assert.assertEquals(getBacklog(topic, subName, partitioned), numMessages);
+
+ tenantManagerAdmin.topics().truncate(topic);
+ Assert.assertEquals(getBacklog(topic, subName, partitioned), 0);
+ superUserAdmin.topics().truncate(topic);
+
+ deleteTopic(topic, partitioned);
+ }
+
+ private long getBacklog(String topic, String subName, boolean partitioned) throws Exception {
+ if (partitioned) {
+ return superUserAdmin.topics().getPartitionedStats(topic, false)
+ .getSubscriptions().get(subName).getMsgBacklog();
+ }
+ return superUserAdmin.topics().getStats(topic).getSubscriptions().get(subName).getMsgBacklog();
+ }
+
@Test
@SneakyThrows
public void testReplicationClusters() {