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/09/29 10:31:20 UTC
(pulsar) branch master updated: [improve][broker] Allow disabling scalable topics when upgrading to 5.x (#26740)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 737c3f86168f [improve][broker] Allow disabling scalable topics when upgrading to 5.x (#26740)
737c3f86168f is described below
commit 737c3f86168f1c8e7f181389a53d694797f948f4
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Sep 29 13:31:13 2026 +0300
[improve][broker] Allow disabling scalable topics when upgrading to 5.x (#26740)
---
conf/broker.conf | 6 ++
conf/standalone.conf | 6 ++
.../apache/pulsar/broker/ServiceConfiguration.java | 6 +-
.../pulsar/broker/resources/PulsarResources.java | 8 +-
.../org/apache/pulsar/broker/PulsarService.java | 4 +-
.../pulsar/broker/lookup/TopicLookupBase.java | 18 ++++-
.../pulsar/broker/service/BrokerService.java | 6 +-
.../apache/pulsar/broker/service/ServerCnx.java | 36 ++++++++-
.../org/apache/pulsar/broker/web/WebService.java | 5 ++
.../service/ScalableTopicBinaryAuthZTest.java | 93 ++++++++++++++++++++++
.../client/api/v5/V5TransactionRecoveryTest.java | 77 ++++++++++++++++++
.../transaction/WatchTcAssignmentsDiscovery.java | 6 +-
.../apache/pulsar/client/impl/ClientCnxTest.java | 20 +++++
.../WatchTcAssignmentsDiscoveryTest.java | 58 ++++++++++++++
14 files changed, 336 insertions(+), 13 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 64a9819537dd..fe7f84c951c3 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -853,6 +853,12 @@ topicFencingTimeoutSeconds=0
# Setting it to true means the broker only sends the list of matching topics to the clients.
enableBrokerSideSubscriptionPatternEvaluation=true
+# Enable scalable topics (PIP-460). Requires a broker restart when changed.
+# Set to false before migrating from 4.x to opt out of scalable topics and preserve
+# rollback options. Disables scalable-topic services, admin APIs, and client access.
+# Existing scalable-topic data is retained but cannot be loaded while disabled.
+scalableTopicsEnabled=true
+
# The maximum length of patterns that a broker evaluates.
# Longer patterns are rejected to avoid patterns that are crafted to overload the broker.
subscriptionPatternMaxLength=50
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 7e48b6853e2f..b9e8facabe8d 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -464,6 +464,12 @@ maxNumPartitionsPerPartitionedTopic=0
# if true the broker only sends the list of matching topics to the clients.
enableBrokerSideSubscriptionPatternEvaluation=true
+# Enable scalable topics (PIP-460). Requires a broker restart when changed.
+# Set to false before migrating from 4.x to opt out of scalable topics and preserve
+# rollback options. Disables scalable-topic services, admin APIs, and client access.
+# Existing scalable-topic data is retained but cannot be loaded while disabled.
+scalableTopicsEnabled=true
+
# The maximum length of patterns that a broker evaluates.
# Longer patterns are rejected to avoid patterns that are crafted to overload the broker.
subscriptionPatternMaxLength=50
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index a5aff1d767a8..49836a82f534 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -1414,7 +1414,11 @@ public class ServiceConfiguration implements PulsarConfiguration {
category = CATEGORY_POLICIES,
doc = "Enables the scalable-topics V5 API on this broker. When disabled, "
+ "the broker advertises supports_scalable_topics=false in CommandConnected "
- + "feature flags and rejects scalable-topic commands from clients."
+ + "feature flags, rejects scalable-topic commands and topic/segment lookups and loads, "
+ + "and does not start scalable-topic services or expose the scalable-topic admin API. "
+ + "Disable before migrating from 4.x to preserve the option to roll back without using "
+ + "scalable topics. Existing scalable-topic data is retained but inaccessible while disabled. "
+ + "Changing this setting requires a broker restart."
)
private boolean scalableTopicsEnabled = true;
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
index db5a8db549c7..1ae32476b7ee 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
@@ -69,6 +69,11 @@ public class PulsarResources {
public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore,
int operationTimeoutSec, Executor executor) {
+ this(localMetadataStore, configurationMetadataStore, operationTimeoutSec, executor, true);
+ }
+
+ public PulsarResources(MetadataStoreExtended localMetadataStore, MetadataStore configurationMetadataStore,
+ int operationTimeoutSec, Executor executor, boolean scalableTopicsEnabled) {
if (configurationMetadataStore != null) {
tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec);
clusterResources = new ClusterResources(localMetadataStore, configurationMetadataStore,
@@ -89,7 +94,8 @@ public class PulsarResources {
bookieResources = new BookieResources(localMetadataStore, operationTimeoutSec);
topicResources = new TopicResources(localMetadataStore);
loadBalanceResources = new LoadBalanceResources(localMetadataStore, operationTimeoutSec);
- scalableTopicResources = new ScalableTopicResources(localMetadataStore, operationTimeoutSec);
+ scalableTopicResources = scalableTopicsEnabled
+ ? new ScalableTopicResources(localMetadataStore, operationTimeoutSec) : null;
} else {
dynamicConfigResources = null;
localPolicies = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 83c05d02e02b..7e2115cf6f4e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -1070,7 +1070,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
.newProvider(config.getTransactionMetadataStoreProviderClassName()), this,
transactionBufferClient, transactionTimer);
- if (config.isTransactionCoordinatorScalableTopicsEnabled()) {
+ if (config.isScalableTopicsEnabled() && config.isTransactionCoordinatorScalableTopicsEnabled()) {
transactionCoordinatorV5 = new TransactionCoordinatorV5(this);
transactionCoordinatorV5.start();
}
@@ -1213,7 +1213,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
@VisibleForTesting
protected PulsarResources newPulsarResources() {
PulsarResources pulsarResources = new PulsarResources(localMetadataStore, configurationMetadataStore,
- config.getMetadataStoreOperationTimeoutSeconds(), getExecutor());
+ config.getMetadataStoreOperationTimeoutSeconds(), getExecutor(), config.isScalableTopicsEnabled());
pulsarResources.getClusterResources().getStore().registerListener(this::handleDeleteCluster);
return pulsarResources;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
index adcb40efe000..faa066db6e3d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java
@@ -60,6 +60,11 @@ public class TopicLookupBase extends PulsarWebResource {
protected CompletableFuture<LookupData> internalLookupTopicAsync(final TopicName topicName, boolean authoritative,
String listenerName) {
+ if (!pulsar().getConfiguration().isScalableTopicsEnabled()
+ && (topicName.isScalable() || topicName.isSegment())) {
+ return FutureUtil.failedFuture(new RestException(Response.Status.PRECONDITION_FAILED,
+ "Scalable topics are disabled on this broker"));
+ }
Semaphore lookupRequestSemaphore = pulsar().getBrokerService().getLookupRequestSemaphore();
if (!lookupRequestSemaphore.tryAcquire()) {
log.warn().attr("topic", topicName)
@@ -189,6 +194,11 @@ public class TopicLookupBase extends PulsarWebResource {
long requestId, final String advertisedListenerName,
Map<String, String> properties) {
+ if (!pulsarService.getConfiguration().isScalableTopicsEnabled()
+ && (topicName.isScalable() || topicName.isSegment())) {
+ return CompletableFuture.completedFuture(newLookupErrorResponse(ServerError.InvalidTopicName,
+ "Scalable topics are disabled on this broker", requestId));
+ }
final CompletableFuture<ByteBuf> validationFuture = new CompletableFuture<>();
final CompletableFuture<ByteBuf> lookupfuture = new CompletableFuture<>();
@@ -368,8 +378,14 @@ public class TopicLookupBase extends PulsarWebResource {
protected TopicName getTopicName(String topicDomain, String tenant, String namespace,
@Encoded String encodedTopic) {
+ TopicDomain domain = TopicDomain.getEnum(topicDomain);
+ if (!pulsar().getConfiguration().isScalableTopicsEnabled()
+ && (domain == TopicDomain.topic || domain == TopicDomain.segment)) {
+ throw new RestException(Response.Status.PRECONDITION_FAILED,
+ "Scalable topics are disabled on this broker");
+ }
String decodedName = Codec.decode(encodedTopic);
- return TopicName.get(TopicDomain.getEnum(topicDomain).value(), tenant, namespace, decodedName);
+ return TopicName.get(domain.value(), tenant, namespace, decodedName);
}
private static boolean shouldRedirectThroughServiceUrl(ServiceConfiguration conf, LookupData lookupData) {
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 cead1b20b141..6a06be0e1b38 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
@@ -646,7 +646,7 @@ public class BrokerService implements Closeable {
// Initialize scalable topic service
var scalableTopicResources = pulsar.getPulsarResources().getScalableTopicResources();
- if (scalableTopicResources != null) {
+ if (pulsar.getConfiguration().isScalableTopicsEnabled() && scalableTopicResources != null) {
this.scalableTopicService = new org.apache.pulsar.broker.service.scalable.ScalableTopicService(
this, scalableTopicResources, pulsar.getCoordinationService());
this.scalableTopicService.start();
@@ -1333,6 +1333,10 @@ public class BrokerService implements Closeable {
*/
public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, boolean createIfMissing,
@Nullable Map<String, String> properties) {
+ if (!pulsar.getConfiguration().isScalableTopicsEnabled()
+ && (topicName.isScalable() || topicName.isSegment())) {
+ return FutureUtil.failedFuture(new NotAllowedException("Scalable topics are disabled on this broker"));
+ }
try {
// If topic future exists in the cache returned directly regardless of whether it fails or timeout.
CompletableFuture<Optional<Topic>> tp = topics.get(topicName.toString());
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index c4a50ffea79b..da87ddb678b1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1020,6 +1020,12 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
checkArgument(state == State.Connected);
long watchId = cmd.getWatchId();
log.debug().attr("watchId", watchId).log("Received WatchScalableTopicsClose");
+ if (!scalableTopicsEnabled) {
+ writeAndFlush(Commands.newWatchScalableTopicsError(watchId, ServerError.NotAllowedError,
+ "Scalable topics are disabled on this broker"));
+ return;
+ }
+
var session = scalableTopicsWatchers.remove(watchId);
if (session != null) {
session.close();
@@ -1035,7 +1041,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
final long watchId = cmd.getWatchId();
log.debug().attr("watchId", watchId).log("Received WatchTcAssignments");
- if (!service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
+ if (!scalableTopicsEnabled
+ || !service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) {
ctx.writeAndFlush(Commands.newWatchTcAssignmentsError(watchId, ServerError.NotAllowedError,
"Scalable-topics transaction coordinator is disabled on this broker"));
return;
@@ -1091,6 +1098,12 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
checkArgument(state == State.Connected);
long watchId = cmd.getWatchId();
log.debug().attr("watchId", watchId).log("Received WatchTcAssignmentsClose");
+ if (!scalableTopicsEnabled) {
+ writeAndFlush(Commands.newWatchTcAssignmentsError(watchId, ServerError.NotAllowedError,
+ "Scalable topics are disabled on this broker"));
+ return;
+ }
+
closeQuietly(tcAssignmentWatchers.remove(watchId));
}
@@ -1116,6 +1129,11 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
checkArgument(state == State.Connected);
final long sessionId = commandScalableTopicClose.getSessionId();
+ if (!scalableTopicsEnabled) {
+ writeAndFlush(Commands.newScalableTopicError(sessionId, ServerError.NotAllowedError,
+ "Scalable topics are disabled on this broker"));
+ return;
+ }
log.debug().attr("sessionId", sessionId).log("Received ScalableTopicClose");
@@ -1238,6 +1256,11 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
checkArgument(state == State.Connected);
final long requestId = commandScalableTopicUnsubscribe.getRequestId();
final long consumerId = commandScalableTopicUnsubscribe.getConsumerId();
+ if (!scalableTopicsEnabled) {
+ writeAndFlush(Commands.newError(requestId, ServerError.NotAllowedError,
+ "Scalable topics are disabled on this broker"));
+ return;
+ }
// The lookup is scoped to this connection's own registrations, so a client can only
// unregister sessions it created here — no further authorization is needed.
@@ -1518,7 +1541,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}
writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, enableTopicListWatcher,
scalableTopicsEnabled,
- service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()));
+ scalableTopicsEnabled
+ && service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()));
state = State.Connected;
service.getPulsarStats().recordConnectionCreateSuccess();
log.debug()
@@ -3706,7 +3730,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
* without the flag always go to the legacy coordinator, so v4 and v5 clients coexist.
*/
private boolean isScalableTcAvailable() {
- return service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()
+ return scalableTopicsEnabled && service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()
&& service.getPulsar().getTransactionCoordinatorV5() != null;
}
@@ -4657,7 +4681,11 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
private TopicName validateTopicName(String topic, long requestId, Object requestCommand) {
try {
- return TopicName.get(topic);
+ TopicName topicName = TopicName.get(topic);
+ if (!scalableTopicsEnabled && (topicName.isScalable() || topicName.isSegment())) {
+ throw new IllegalArgumentException("Scalable topics are disabled on this broker");
+ }
+ return topicName;
} catch (Throwable t) {
log.debug()
.attr("topic", topic)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
index 7c8e523dac5b..0e0d5c0193fd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java
@@ -43,6 +43,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.admin.v2.ScalableTopics;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.BrokerInterceptors;
import org.apache.pulsar.broker.tls.DefaultBrokerTlsFactory;
@@ -228,6 +229,10 @@ public class WebService implements AutoCloseable {
for (String javaPackage : javaPackages) {
config.packages(false, javaPackage);
}
+ if (!pulsar.getConfiguration().isScalableTopicsEnabled()) {
+ config = new ResourceConfig(config.getClasses().stream()
+ .filter(resource -> resource != ScalableTopics.class).toArray(Class<?>[]::new));
+ }
addResourceServlet(basePath, requiresAuthentication, attributeMap, config, useSharedJsonMapperProvider);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java
index bc0c680ea26d..74ac52e6b090 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ScalableTopicBinaryAuthZTest.java
@@ -19,6 +19,8 @@
package org.apache.pulsar.broker.service;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgsRecordingInvocations;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
@@ -33,9 +35,12 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockAuthenticationProvider;
@@ -48,11 +53,14 @@ import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.scalable.ScalableTopicService;
import org.apache.pulsar.broker.service.utils.ClientChannelHelper;
import org.apache.pulsar.broker.testcontext.PulsarTestContext;
+import org.apache.pulsar.common.api.proto.BaseCommand;
import org.apache.pulsar.common.api.proto.CommandConnected;
import org.apache.pulsar.common.api.proto.CommandScalableTopicSubscribeResponse;
import org.apache.pulsar.common.api.proto.CommandScalableTopicUpdate;
+import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.ScalableConsumerType;
import org.apache.pulsar.common.api.proto.ServerError;
+import org.apache.pulsar.common.api.proto.TxnAction;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TopicOperation;
import org.apache.pulsar.common.protocol.Commands;
@@ -256,6 +264,86 @@ public class ScalableTopicBinaryAuthZTest {
verify(authorizationService, never()).allowTopicOperationAsync(any(), any(), any(), any());
}
+ @Test
+ public void testDisabledCommandsReturnErrors() throws Exception {
+ svcConfig.setScalableTopicsEnabled(false);
+ svcConfig.setTransactionCoordinatorEnabled(true);
+ svcConfig.setTransactionCoordinatorScalableTopicsEnabled(true);
+ configureAuthAndConnect(AUTHORIZED_ROLE);
+
+ List<Supplier<ByteBuf>> commands = List.of(
+ () -> Commands.newScalableTopicLookup(1, TOPIC),
+ () -> Commands.newScalableTopicClose(1),
+ () -> Commands.newScalableTopicSubscribe(1, TOPIC, SUBSCRIPTION,
+ "consumer", 1, ScalableConsumerType.STREAM),
+ () -> Commands.newScalableTopicUnsubscribe(1, 1),
+ () -> Commands.newWatchScalableTopics(1, "public/default", null, null),
+ () -> Commands.newWatchScalableTopicsClose(1),
+ () -> Commands.newWatchTcAssignments(1),
+ () -> Commands.newWatchTcAssignmentsClose(1),
+ () -> Commands.newTcClientConnectRequest(0, 1, true),
+ () -> Commands.newTxn(0, 1, 10000, true),
+ () -> Commands.newAddPartitionToTxn(1, 0, 0, List.of(TOPIC), true),
+ () -> Commands.newAddSubscriptionToTxn(1, 0, 0, List.of(), true),
+ () -> Commands.serializeWithSize(Commands.newEndTxn(1, 0, 0, TxnAction.COMMIT, true)));
+ for (Supplier<ByteBuf> command : commands) {
+ channel.writeInbound(command.get());
+ BaseCommand response = readProtocolResponse();
+ ServerError error = switch (response.getType()) {
+ case SCALABLE_TOPIC_UPDATE -> response.getScalableTopicUpdate().getError();
+ case SCALABLE_TOPIC_SUBSCRIBE_RESPONSE -> response.getScalableTopicSubscribeResponse().getError();
+ case WATCH_SCALABLE_TOPICS_UPDATE -> response.getWatchScalableTopicsUpdate().getError();
+ case WATCH_TC_ASSIGNMENTS_UPDATE -> response.getWatchTcAssignmentsUpdate().getError();
+ case TC_CLIENT_CONNECT_RESPONSE -> response.getTcClientConnectResponse().getError();
+ case NEW_TXN_RESPONSE -> response.getNewTxnResponse().getError();
+ case ADD_PARTITION_TO_TXN_RESPONSE -> response.getAddPartitionToTxnResponse().getError();
+ case ADD_SUBSCRIPTION_TO_TXN_RESPONSE -> response.getAddSubscriptionToTxnResponse().getError();
+ case END_TXN_RESPONSE -> response.getEndTxnResponse().getError();
+ case ERROR -> response.getError().getError();
+ default -> throw new AssertionError("Unexpected response: " + response.getType());
+ };
+ assertThat(error).as("error for %s", response.getType()).isEqualTo(ServerError.NotAllowedError);
+ }
+ verify(authorizationService, never()).allowTopicOperationAsync(any(), any(), any(), any());
+ }
+
+ @Test
+ public void testDisabledTopicDomainsRejectedByLookupAndPartitionMetadata() throws Exception {
+ svcConfig.setScalableTopicsEnabled(false);
+ configureAuthAndConnect(AUTHORIZED_ROLE);
+ for (String topic : List.of("topic://public/default/test", "segment://public/default/test/0000-ffff-0")) {
+ channel.writeInbound(Commands.newLookup(topic, false, 1));
+ var lookup = readProtocolResponse().getLookupTopicResponse();
+ assertThat(lookup.getError()).isEqualTo(ServerError.InvalidTopicName);
+ assertThat(lookup.getMessage()).contains("Scalable topics are disabled");
+ channel.writeInbound(Commands.newPartitionMetadataRequest(topic, 2, true));
+ assertThat(readProtocolResponse().getPartitionMetadataResponse().getError())
+ .isEqualTo(ServerError.InvalidTopicName);
+ channel.writeInbound(Commands.newProducer(topic, 1, 3, "producer", Map.of(), false));
+ assertThat(readProtocolResponse().getError().getError()).isEqualTo(ServerError.InvalidTopicName);
+ channel.writeInbound(Commands.newSubscribe(topic, "subscription", 1, 4,
+ SubType.Shared, 0, "consumer", 0));
+ assertThat(readProtocolResponse().getError().getError()).isEqualTo(ServerError.InvalidTopicName);
+ }
+ }
+
+ private BaseCommand readProtocolResponse() {
+ await().until(() -> {
+ channel.runPendingTasks();
+ return !channel.outboundMessages().isEmpty();
+ });
+ ByteBuf frame = channel.readOutbound();
+ try {
+ frame.readInt(); // frame size
+ int commandSize = frame.readInt();
+ BaseCommand command = new BaseCommand();
+ command.parseFrom(frame, commandSize);
+ return new BaseCommand().copyFrom(command);
+ } finally {
+ frame.release();
+ }
+ }
+
private void configureAuthAndConnect(String role) throws Exception {
AuthenticationService authenticationService = mock(AuthenticationService.class);
AuthenticationProvider authenticationProvider = new MockAuthenticationProvider();
@@ -276,6 +364,11 @@ public class ScalableTopicBinaryAuthZTest {
Object response = readResponse();
assertTrue(response instanceof CommandConnected, "Expected CommandConnected, got " + response);
+ var flags = ((CommandConnected) response).getFeatureFlags();
+ assertThat(flags.isSupportsScalableTopics()).isEqualTo(svcConfig.isScalableTopicsEnabled());
+ if (!svcConfig.isScalableTopicsEnabled()) {
+ assertThat(flags.isSupportsTcMetadataDiscovery()).isFalse();
+ }
}
private void resetChannel() throws Exception {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java
index 64fd9e885449..6c19c3fc8b1a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionRecoveryTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.client.api.v5;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
@@ -29,14 +31,20 @@ import java.util.UUID;
import lombok.Cleanup;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
import org.apache.pulsar.client.api.v5.schema.Schema;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.SystemTopicNames;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -101,6 +109,75 @@ public class V5TransactionRecoveryTest extends MockedPulsarServiceBaseTest {
return name;
}
+ @Test
+ public void testDisableScalableTopicsAcrossRestart() throws Exception {
+ assertThat(conf.isScalableTopicsEnabled()).isTrue();
+ String topic = newScalableTopic(1);
+ try (var producer = v5Client.newProducer(Schema.string()).topic(topic).create()) {
+ producer.newMessage().value("before-disable").send();
+ }
+ var metadata = admin.scalableTopics().getMetadata(topic);
+ var segmentInfo = metadata.getSegments().values().iterator().next();
+ String segment = SegmentTopicName.fromParent(TopicName.get(topic),
+ new HashRange(segmentInfo.getHashRange().getStart(), segmentInfo.getHashRange().getEnd()),
+ segmentInfo.getSegmentId()).toString();
+ assertThat(pulsar.getBrokerService().getTopicIfExists(segment).get()).isPresent();
+
+ v5Client.close();
+ restartBroker(config -> config.setScalableTopicsEnabled(false));
+ assertThatThrownBy(() -> newV5Client(Duration.ofSeconds(10)))
+ .hasMessageContaining("Broker does not support scalable-topics transactions");
+ v5Client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build();
+ assertThat(pulsar.getPulsarResources().getScalableTopicResources()).isNull();
+ assertThat(pulsar.getBrokerService().getScalableTopicService()).isNull();
+ // The separate transaction setting must not override the feature-wide switch.
+ assertThat(conf.isTransactionCoordinatorScalableTopicsEnabled()).isTrue();
+ assertThat(pulsar.getTransactionCoordinatorV5()).isNull();
+ assertThatThrownBy(() -> admin.scalableTopics().getMetadata(topic))
+ .isInstanceOf(PulsarAdminException.NotFoundException.class);
+ assertThatThrownBy(() -> admin.scalableTopics().createScalableTopic(topic + "-new", 1))
+ .isInstanceOf(PulsarAdminException.NotFoundException.class);
+ for (String name : Set.of(topic, segment)) {
+ assertThatThrownBy(() -> pulsar.getBrokerService().getTopicIfExists(name).get())
+ .hasCauseInstanceOf(BrokerServiceException.NotAllowedException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ assertThatThrownBy(() -> admin.lookups().lookupTopic(name))
+ .isInstanceOf(PulsarAdminException.PreconditionFailedException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ }
+ assertThatThrownBy(() -> v5Client.newProducer(Schema.string()).topic(topic).create())
+ .hasRootCauseInstanceOf(PulsarClientException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ assertThatThrownBy(() -> v5Client.newQueueConsumer(Schema.string()).topic(topic)
+ .subscriptionName("queue").subscribe())
+ .hasRootCauseInstanceOf(PulsarClientException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ assertThatThrownBy(() -> v5Client.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribe())
+ .hasRootCauseInstanceOf(PulsarClientException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ assertThatThrownBy(() -> v5Client.newCheckpointConsumer(Schema.string()).topic(topic)
+ .consumerGroup("checkpoint").create())
+ .hasRootCauseInstanceOf(PulsarClientException.class)
+ .hasMessageContaining("Scalable topics are disabled");
+ // Ordinary topics remain available to the classic client while scalable topics are disabled.
+ String regular = topic.replace("topic://", "persistent://") + "-regular";
+ try (var producer = pulsarClient.newProducer().topic(regular).create()) {
+ producer.send(new byte[] {1});
+ }
+
+ v5Client.close();
+ restartBroker(config -> config.setScalableTopicsEnabled(true));
+ v5Client = newV5Client(Duration.ofSeconds(10));
+ assertThat(pulsar.getPulsarResources().getScalableTopicResources()).isNotNull();
+ assertThat(pulsar.getBrokerService().getScalableTopicService()).isNotNull();
+ assertThat(admin.scalableTopics().getMetadata(topic).getSegments().keySet())
+ .isEqualTo(metadata.getSegments().keySet());
+ try (var consumer = subscribe(topic, "after-enable")) {
+ assertThat(consumer.receive(Duration.ofSeconds(10)).value()).isEqualTo("before-disable");
+ }
+ }
+
private QueueConsumer<String> subscribe(String topic, String sub) throws Exception {
return v5Client.newQueueConsumer(Schema.string())
.topic(topic)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscovery.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscovery.java
index e1d8609edf1a..8c0e11168db2 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscovery.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscovery.java
@@ -104,11 +104,11 @@ class WatchTcAssignmentsDiscovery implements TcDiscovery, ClientCnx.TcAssignment
}
if (!newCnx.isSupportsTcMetadataDiscovery()) {
// The broker we landed on doesn't support the watch. On the very first open this is a
- // hard failure (the caller chose this strategy on a probe that said it was supported);
+ // hard failure (the V5 client requires the metadata-store coordinator);
// after that, it's likely transient — config drift or we hit a different/old broker —
// so reconnect to find a supporting broker rather than freezing on the last snapshot.
- onAttachFailure(new PulsarClientException(
- "Broker does not support metadata-store TC discovery"));
+ onAttachFailure(new PulsarClientException.NotSupportedException(
+ "Broker does not support scalable-topics transactions (metadata-store TC discovery)"));
return;
}
this.cnx = newCnx;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
index 71590e7ff4b8..bfc1983f9db2 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxTest.java
@@ -332,6 +332,26 @@ public class ClientCnxTest {
assertFalse(cnx.idleCheck());
}
+ @Test
+ public void testScalableTopicSupportRequiresExplicitFeatureFlag() {
+ // Older brokers either omit feature flags entirely or advertise only older flags.
+ for (int mode = 0; mode < 4; mode++) {
+ CommandConnected connected = new CommandConnected()
+ .setServerVersion("test")
+ .setProtocolVersion(21);
+ if (mode == 1) {
+ connected.setFeatureFlags().setSupportsTopicWatchers(true);
+ } else if (mode >= 2) {
+ connected.setFeatureFlags().setSupportsScalableTopics(mode == 3);
+ }
+ boolean expectedSupport = mode == 3;
+ withConnection("testScalableTopicSupportRequiresExplicitFeatureFlag", cnx -> {
+ cnx.handleConnected(connected);
+ assertThat(cnx.isSupportsScalableTopics()).isEqualTo(expectedSupport);
+ });
+ }
+ }
+
@Test
public void testNoWatchersWhenNoServerSupport() {
withConnection("testNoWatchersWhenNoServerSupport", cnx -> {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscoveryTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscoveryTest.java
new file mode 100644
index 000000000000..5e4efbdc3065
--- /dev/null
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/transaction/WatchTcAssignmentsDiscoveryTest.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.transaction;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.testng.annotations.Test;
+
+public class WatchTcAssignmentsDiscoveryTest {
+
+ @Test
+ public void testUnsupportedBrokerFailsWithoutRetrying() {
+ PulsarClientImpl client = mock(PulsarClientImpl.class);
+ ClientCnx cnx = mock(ClientCnx.class);
+ when(client.getConfiguration()).thenReturn(new ClientConfigurationData());
+ when(client.getAnyBrokerProxyConnection()).thenReturn(CompletableFuture.completedFuture(cnx));
+ WatchTcAssignmentsDiscovery discovery = new WatchTcAssignmentsDiscovery(client);
+ try {
+ CompletableFuture<Void> start = discovery.start();
+ assertThat(start).isCompletedExceptionally();
+ assertThatThrownBy(start::join)
+ .hasCauseInstanceOf(PulsarClientException.NotSupportedException.class)
+ .hasMessageContaining("Broker does not support scalable-topics transactions");
+ verify(client, never()).timer();
+ verify(cnx, never()).registerTcAssignmentsWatcher(anyLong(), any());
+ verify(cnx, never()).ctx();
+ } finally {
+ discovery.close();
+ }
+ }
+}