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/21 19:45:34 UTC
(pulsar) 04/21: [fix][broker] Reject unsupported subscription-only backlog requests (#26906)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 53ada91eaf5538bdb6ff80224b0d32462c2b0518
Author: sinan liu <li...@gmail.com>
AuthorDate: Wed Oct 21 07:50:55 2026 +0800
[fix][broker] Reject unsupported subscription-only backlog requests (#26906)
(cherry picked from commit 908bfe95a04ddadb778cd0cfbfcc512c73a5e08a)
---
.../apache/pulsar/broker/admin/v2/Namespaces.java | 13 +++
.../pulsar/broker/admin/NamespaceAuthZTest.java | 101 +++++++++++++++++++++
.../client/admin/internal/NamespacesImpl.java | 8 +-
3 files changed, 118 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index f68bb3efe0ec..e2e9e118c380 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -1946,6 +1946,19 @@ public class Namespaces extends NamespacesBase {
});
}
+ @POST
+ @Path("/{tenant}/{namespace}/{bundle}/clearSubscriptionBacklog/{subscription}")
+ @Operation(hidden = true)
+ public void clearNamespaceBundleSubscriptionBacklog(@Suspended final AsyncResponse asyncResponse,
+ @PathParam("tenant") String tenant, @PathParam("namespace") String namespace,
+ @PathParam("subscription") String subscription, @PathParam("bundle") String bundleRange,
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ // Keep the old query parameter for existing senders, but use a distinct route for new internal requests.
+ // An owner without this route returns 404 instead of silently dropping the subscription-only restriction.
+ clearNamespaceBundleBacklogForSubscription(asyncResponse, tenant, namespace, subscription, bundleRange,
+ authoritative, true);
+ }
+
@POST
@Path("/{tenant}/{namespace}/unsubscribe/{subscription}")
@Operation(summary = "Unsubscribes the given subscription on all topics on a namespace.")
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespaceAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespaceAuthZTest.java
index 792e970e6ae9..ef6e85574c2f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespaceAuthZTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespaceAuthZTest.java
@@ -26,8 +26,11 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import com.google.common.collect.Sets;
+import com.sun.net.httpserver.HttpServer;
import io.jsonwebtoken.Jwts;
import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
@@ -43,6 +46,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;
import lombok.Cleanup;
@@ -1142,6 +1146,103 @@ public class NamespaceAuthZTest extends MockedPulsarStandalone {
superUserAdmin.clusters().deleteCluster(remoteCluster);
}
+ @Test
+ public void testNamespaceSubscriptionForwardingToLegacyBundleOwner() throws Exception {
+ final String random = UUID.randomUUID().toString();
+ final String tenant = "tenant-" + random;
+ final String namespace = tenant + "/ns";
+ final String topic = "persistent://" + namespace + "/topic";
+ final String remoteCluster = "remote-" + random;
+ final String localCluster = getServiceConfiguration().getClusterName();
+ final String subject = "consumer-" + random;
+ final String token = Jwts.builder().claim("sub", subject).signWith(SECRET_KEY).compact();
+ final String bundle = "0x00000000_0xffffffff";
+ final int numMessages = 5;
+ superUserAdmin.clusters().createCluster(remoteCluster, ClusterData.builder()
+ .serviceUrl("http://127.0.0.1:1").brokerServiceUrl("pulsar://127.0.0.1:1").build());
+ superUserAdmin.tenants().createTenant(tenant, TenantInfo.builder()
+ .adminRoles(Set.of(TENANT_ADMIN_SUBJECT)).allowedClusters(Set.of(localCluster, remoteCluster)).build());
+ superUserAdmin.namespaces().createNamespace(namespace, 1);
+ superUserAdmin.topics().createNonPartitionedTopic(topic);
+ superUserAdmin.namespaces().setNamespaceReplicationClusters(namespace, Set.of(localCluster, remoteCluster),
+ false);
+ superUserAdmin.namespaces().grantPermissionOnNamespace(namespace, subject, Set.of(AuthAction.consume));
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(false).create();
+ @Cleanup
+ PulsarAdmin subAdmin = PulsarAdmin.builder().serviceHttpUrl(getPulsarService().getWebServiceAddress())
+ .authentication(new AuthenticationToken(token)).build();
+ @Cleanup
+ HttpClient httpClient = HttpClient.newHttpClient();
+ HttpServer legacyOwner = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ AtomicInteger legacyBundleRequests = new AtomicInteger();
+ // Model the old owner's REST protocol: only the old route exists, and unknown query parameters are ignored.
+ // Forward to the real broker without subscriptionOnly, preserving the internal client's credentials. Its
+ // admin path executes the same cursor fallback as the old owner, using real replication cursors.
+ legacyOwner.createContext("/admin/v2/namespaces/" + namespace + "/" + bundle
+ + "/clearBacklog/" + remoteCluster, exchange -> {
+ legacyBundleRequests.incrementAndGet();
+ HttpRequest request = HttpRequest.newBuilder(URI.create(getPulsarService().getWebServiceAddress()
+ + exchange.getRequestURI().getRawPath()))
+ .header("Authorization", exchange.getRequestHeaders().getFirst("Authorization"))
+ .header("Content-Type", "application/json")
+ .POST(HttpRequest.BodyPublishers.ofByteArray(exchange.getRequestBody().readAllBytes())).build();
+ try {
+ HttpResponse<byte[]> response = httpClient.send(request, HttpResponse.BodyHandlers.ofByteArray());
+ exchange.sendResponseHeaders(response.statusCode(), response.body().length == 0
+ ? -1 : response.body().length);
+ exchange.getResponseBody().write(response.body());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ } finally {
+ exchange.close();
+ }
+ });
+ legacyOwner.start();
+ @Cleanup
+ PulsarAdmin internalAdmin = PulsarAdmin.builder()
+ .serviceHttpUrl("http://127.0.0.1:" + legacyOwner.getAddress().getPort())
+ .authentication(getServiceConfiguration().getBrokerClientAuthenticationPlugin(),
+ getServiceConfiguration().getBrokerClientAuthenticationParameters()).build();
+ try {
+ Awaitility.await().untilAsserted(() -> assertNotNull(
+ superUserAdmin.topics().getStats(topic).getReplication().get(remoteCluster)));
+ sendMessages(producer, numMessages);
+ Awaitility.await().untilAsserted(() -> assertEquals(getReplicationBacklog(topic, remoteCluster),
+ numMessages));
+ // Confirm that the legacy route really resolves the bare cluster name to the replication cursor.
+ internalAdmin.namespaces().clearNamespaceBundleBacklogForSubscription(namespace, bundle, remoteCluster);
+ Awaitility.await().untilAsserted(() -> assertEquals(getReplicationBacklog(topic, remoteCluster), 0));
+ sendMessages(producer, numMessages);
+ Awaitility.await().untilAsserted(() -> assertEquals(getReplicationBacklog(topic, remoteCluster),
+ numMessages));
+
+ // Only replace the destination of the real internal HTTP client; authorization and cursor operations
+ // still run through the production namespace REST entry point and the real broker.
+ Mockito.doReturn(internalAdmin).when(getPulsarService()).getAdminClient();
+ PulsarAdminException failure = null;
+ try {
+ subAdmin.namespaces().clearNamespaceBacklogForSubscription(namespace, remoteCluster);
+ } catch (PulsarAdminException e) {
+ failure = e;
+ }
+ assertEquals(getReplicationBacklog(topic, remoteCluster), numMessages,
+ "A subscription-only request must not clear the legacy owner's replication cursor");
+ assertTrue(failure instanceof PulsarAdminException.NotFoundException,
+ "An owner without the subscription-only route must reject the request");
+ assertEquals(legacyBundleRequests.get(), 1, "The restricted request must not use the legacy route");
+ } finally {
+ Mockito.doCallRealMethod().when(getPulsarService()).getAdminClient();
+ legacyOwner.stop(0);
+ producer.close();
+ superUserAdmin.namespaces().setNamespaceReplicationClusters(namespace, Set.of(localCluster), false);
+ deleteNamespaceWithRetry(namespace, true, superUserAdmin);
+ superUserAdmin.tenants().deleteTenant(tenant);
+ superUserAdmin.clusters().deleteCluster(remoteCluster);
+ }
+ }
+
@Test
public void testNamespaceClearBacklogForSubscriptionWithTopicLevelReplication() throws Exception {
final String random = UUID.randomUUID().toString();
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
index 90cc63aa02b5..82734a7c5291 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java
@@ -1212,14 +1212,14 @@ public class NamespacesImpl extends BaseResource implements Namespaces {
/**
* Used by the broker when dispatching to namespace bundles; not part of the public
* {@link org.apache.pulsar.client.admin.Namespaces} API.
+ * The subscription-only route is deliberately distinct so that older owners reject it rather than ignore
+ * an unknown query parameter and resolve the name as a replicator cursor.
*/
public CompletableFuture<Void> clearNamespaceBundleBacklogForSubscriptionAsync(String namespace, String bundle,
String subscription, boolean subscriptionOnly) {
NamespaceName ns = NamespaceName.get(namespace);
- WebTarget path = namespacePath(ns, bundle, "clearBacklog", subscription);
- if (subscriptionOnly) {
- path = path.queryParam("subscriptionOnly", true);
- }
+ WebTarget path = namespacePath(ns, bundle,
+ subscriptionOnly ? "clearSubscriptionBacklog" : "clearBacklog", subscription);
return asyncPostRequest(path, Entity.entity("", MediaType.APPLICATION_JSON));
}