You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/13 01:25:20 UTC
(pulsar) branch master updated: [fix][broker] Stop scalable topic controllers from re-electing while the topic is deleted (#26875)
This is an automated email from the ASF dual-hosted git repository.
merlimat 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 9f8e1c7637d1 [fix][broker] Stop scalable topic controllers from re-electing while the topic is deleted (#26875)
9f8e1c7637d1 is described below
commit 9f8e1c7637d19a13258e85f86bc8283db26e2209
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 12 18:25:12 2026 -0700
[fix][broker] Stop scalable topic controllers from re-electing while the topic is deleted (#26875)
---
.../broker/resources/ScalableTopicMetadata.java | 8 ++
.../broker/resources/ScalableTopicResources.java | 11 ++
.../pulsar/broker/admin/v2/ScalableTopics.java | 45 +++++-
.../service/scalable/ScalableTopicController.java | 154 ++++++++++++++++++++
.../service/scalable/ScalableTopicService.java | 20 ++-
.../broker/service/scalable/SegmentLayout.java | 5 +-
.../scalable/ScalableTopicControllerTest.java | 50 +++++++
.../service/scalable/ScalableTopicServiceTest.java | 56 ++++++++
.../broker/service/scalable/SegmentLayoutTest.java | 2 +
.../v5/V5MultiBrokerScalableTopicDeleteTest.java | 155 +++++++++++++++++++++
.../client/api/v5/V5ScalableTopicDeleteTest.java | 118 ++++++++++++++++
11 files changed, 615 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicMetadata.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicMetadata.java
index 228e2c25e651..fec2ed76f3fe 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicMetadata.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicMetadata.java
@@ -58,4 +58,12 @@ public class ScalableTopicMetadata {
* the namespace policy and then the broker configuration apply.
*/
private AutoScalePolicyOverride autoScalePolicy;
+
+ /**
+ * Set while the topic is being deleted, like {@code PartitionedTopicMetadata.deleted}. No
+ * controller elects itself for a topic marked deleted, so the delete can remove the
+ * controller lock along with the rest of the topic's metadata. A delete that fails after
+ * marking the topic leaves it marked, until a retried delete removes it.
+ */
+ private boolean deleted;
}
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicResources.java
index bec03c6029df..ae98693331d5 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ScalableTopicResources.java
@@ -272,6 +272,17 @@ public class ScalableTopicResources extends BaseResources<ScalableTopicMetadata>
return getStore().deleteRecursive(topicPath(tn));
}
+ /**
+ * Mark the topic as being deleted, before its metadata is deleted. See
+ * {@link ScalableTopicMetadata#isDeleted()}.
+ */
+ public CompletableFuture<Void> markScalableTopicDeletedAsync(TopicName tn) {
+ return updateScalableTopicAsync(tn, md -> {
+ md.setDeleted(true);
+ return md;
+ });
+ }
+
public CompletableFuture<Boolean> scalableTopicExistsAsync(TopicName tn) {
return existsAsync(topicPath(tn));
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
index 84c8e0e09fa9..388eacd8e5a9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/ScalableTopics.java
@@ -647,6 +647,8 @@ public class ScalableTopics extends AdminResource {
@Operation(summary = "Delete a scalable topic and all its segments.")
@ApiResponses(value = {
@ApiResponse(responseCode = "204", description = "Scalable topic deleted successfully"),
+ @ApiResponse(responseCode = "307",
+ description = "Current broker is not the controller leader for this topic"),
@ApiResponse(responseCode = "401",
description = "Don't have permission to administrate resources on this tenant"),
@ApiResponse(responseCode = "403", description = "Don't have admin permission on the namespace"),
@@ -665,17 +667,30 @@ public class ScalableTopics extends AdminResource {
validateNamespaceName(tenant, namespace);
TopicName tn = TopicName.get(TopicDomain.topic.value(), namespaceName, encodedTopic);
+ // Served by the controller leader, so the broker holding the elected controller is the
+ // one that stops it below.
validateNamespaceOperationAsync(namespaceName, NamespaceOperation.DELETE_TOPIC)
+ .thenCompose(__ -> redirectToControllerLeaderIfNeeded(tn))
.thenCompose(__ -> resources().getScalableTopicMetadataAsync(tn))
.thenCompose(optMd -> {
if (optMd.isEmpty()) {
throw new RestException(Response.Status.NOT_FOUND,
"Scalable topic not found: " + tn);
}
- // Delete metadata first, then best-effort clean up segment topics and the schema
- // they share (deleting a segment topic leaves the schema in place)
- return resources().deleteScalableTopicAsync(tn)
- .thenCompose(__ -> deleteSegmentTopics(tn, optMd.get(), force))
+ // Stop the controller before deleting the metadata: still running, its
+ // leader election re-acquires the lock the delete removes, under the deleted
+ // path, and it keeps serving consumers from the deleted layout. Mark the
+ // topic deleted first, so the controllers following it on other brokers
+ // don't take the lock over once it is released. Then delete metadata, then
+ // best-effort clean up segment topics and the schema they share (deleting a
+ // segment topic leaves the schema in place). The segments are read again
+ // after the release, to include those of a split that was still in flight.
+ return resources().markScalableTopicDeletedAsync(tn)
+ .thenCompose(__ -> releaseController(tn))
+ .thenCompose(__ -> resources().getScalableTopicMetadataAsync(tn, true))
+ .thenCompose(latest -> resources().deleteScalableTopicAsync(tn)
+ .thenCompose(__ -> deleteSegmentTopics(tn, latest.orElse(optMd.get()),
+ force)))
.thenCompose(__ -> pulsar().getBrokerService().deleteSchema(tn)
.exceptionally(ex -> {
log.warn().attr("topic", tn).exceptionMessage(ex)
@@ -689,8 +704,10 @@ public class ScalableTopics extends AdminResource {
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
- log.error().attr("clientAppId", clientAppId()).attr("topic", tn)
- .exception(ex).log("Failed to delete scalable topic");
+ if (!isRedirectException(ex)) {
+ log.error().attr("clientAppId", clientAppId()).attr("topic", tn)
+ .exception(ex).log("Failed to delete scalable topic");
+ }
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
@@ -1221,6 +1238,22 @@ public class ScalableTopics extends AdminResource {
}
}
+ /**
+ * Stop this broker's controller for {@code tn}, if it holds one, closing its leader election.
+ * Best-effort: a failure is logged and the delete goes on.
+ */
+ private CompletableFuture<Void> releaseController(TopicName tn) {
+ ScalableTopicService svc = pulsar().getBrokerService().getScalableTopicService();
+ if (svc == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return svc.releaseController(tn).exceptionally(ex -> {
+ log.warn().attr("topic", tn).exceptionMessage(ex)
+ .log("Failed to release the controller of the scalable topic being deleted");
+ return null;
+ });
+ }
+
/**
* Best-effort delete the underlying topic for every segment in the DAG. Uses the
* segment-aware admin endpoint, which routes to the segment-owning broker via the
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
index fa80cc246636..9e8c9d67fef7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java
@@ -58,8 +58,11 @@ import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.scalable.HashRange;
import org.apache.pulsar.common.scalable.SegmentInfo;
import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.compaction.Compactor;
+import org.apache.pulsar.metadata.api.Notification;
+import org.apache.pulsar.metadata.api.NotificationType;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
import org.apache.pulsar.metadata.api.coordination.LeaderElection;
import org.apache.pulsar.metadata.api.coordination.LeaderElectionState;
@@ -125,6 +128,17 @@ public class ScalableTopicController {
private volatile boolean closed = false;
+ /** Registered when this controller elects, see {@link #onControllerLockDeleted()}. */
+ private final ControllerLockListener controllerLockListener = new ControllerLockListener();
+ /** Registered when this controller elects, see {@link #onTopicRecordDeleted()}. */
+ private final TopicRecordListener topicRecordListener = new TopicRecordListener();
+
+ /** Set while re-electing after the controller lock was deleted, retries included. */
+ private final AtomicBoolean reElecting = new AtomicBoolean(false);
+
+ /** Run once this controller finds its topic deleted. Set by {@link ScalableTopicService}. */
+ private volatile Runnable topicDeletedAction = () -> { };
+
ScalableTopicController(TopicName topicName,
ScalableTopicResources resources,
BrokerService brokerService,
@@ -176,6 +190,102 @@ public class ScalableTopicController {
}
}
+ /**
+ * The controller lock was deleted while this broker follows the leader: the leader released
+ * the controller or lost its metadata session, or the topic is being deleted. A follower's
+ * election is closed once it settles (see {@link #electLeader()}), so it doesn't take the lock
+ * over by itself, even for a topic being deleted, where it would re-create the lock under the
+ * record the delete is about to remove (failing that delete on stores that refuse to delete a
+ * node with children). Re-elect through {@link #initialize()} instead, unless the topic is gone
+ * or marked deleted, retrying with backoff while that fails.
+ */
+ private void onControllerLockDeleted() {
+ if (closed || leaderState != LeaderElectionState.Following
+ || !reElecting.compareAndSet(false, true)) {
+ return;
+ }
+ reElect(Backoff.builder()
+ .initialDelay(Duration.ofMillis(100))
+ .maxBackoff(Duration.ofSeconds(30))
+ .build());
+ }
+
+ private void reElect(Backoff backoff) {
+ resources.getScalableTopicMetadataAsync(topicName, true)
+ .thenCompose(optMd -> {
+ if (closed || optMd.isEmpty() || optMd.get().isDeleted()) {
+ log.debug().log("Controller lock released for a deleted topic, not re-electing");
+ return CompletableFuture.completedFuture(null);
+ }
+ return initialize();
+ })
+ .whenComplete((__, ex) -> {
+ if (ex == null || closed) {
+ reElecting.set(false);
+ return;
+ }
+ long delayMs = backoff.next().toMillis();
+ log.warn().attr("retryInMs", delayMs).exceptionMessage(ex)
+ .log("Failed to re-elect after the controller lock was released");
+ scheduler().schedule(() -> reElect(backoff), delayMs, TimeUnit.MILLISECONDS);
+ });
+ }
+
+ /**
+ * The topic record was deleted. The admin delete stops the controller on the broker serving
+ * it, but other brokers can hold one for the topic too; left running, it keeps serving
+ * consumers from the deleted layout. The event can arrive after the topic was re-created under
+ * the same name, so act only if the record is still gone.
+ */
+ private void onTopicRecordDeleted() {
+ if (closed) {
+ return;
+ }
+ resources.getScalableTopicMetadataAsync(topicName, true)
+ .thenAccept(optMd -> {
+ if (optMd.isEmpty() && !closed) {
+ log.debug().log("Scalable topic deleted, releasing its controller");
+ topicDeletedAction.run();
+ }
+ })
+ .exceptionally(ex -> {
+ log.warn().exceptionMessage(ex).log("Failed to check the record of a deleted topic");
+ return null;
+ });
+ }
+
+ void setTopicDeletedAction(Runnable topicDeletedAction) {
+ this.topicDeletedAction = topicDeletedAction;
+ }
+
+ private final class ControllerLockListener implements ScalableTopicResources.MetadataPathListener {
+ @Override
+ public String getMetadataPath() {
+ return resources.controllerLockPath(topicName);
+ }
+
+ @Override
+ public void onNotification(Notification notification) {
+ if (notification.getType() == NotificationType.Deleted) {
+ onControllerLockDeleted();
+ }
+ }
+ }
+
+ private final class TopicRecordListener implements ScalableTopicResources.MetadataPathListener {
+ @Override
+ public String getMetadataPath() {
+ return resources.topicPath(topicName);
+ }
+
+ @Override
+ public void onNotification(Notification notification) {
+ if (notification.getType() == NotificationType.Deleted) {
+ onTopicRecordDeleted();
+ }
+ }
+ }
+
/**
* Initialize: load current layout from metadata store and attempt to become leader.
*
@@ -192,9 +302,15 @@ public class ScalableTopicController {
return CompletableFuture.failedFuture(
new IllegalStateException("Scalable topic not found: " + topicName));
}
+ if (optMd.get().isDeleted()) {
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("Scalable topic is being deleted: " + topicName));
+ }
this.currentLayout = SegmentLayout.fromMetadata(optMd.get());
return electLeader();
})
+ .thenCompose(__ -> isLeader()
+ ? releaseIfMarkedDeleted() : CompletableFuture.completedFuture(null))
.thenCompose(__ -> {
if (isLeader()) {
seedAutoScaleCooldownsFromLayout();
@@ -207,6 +323,26 @@ public class ScalableTopicController {
});
}
+ /**
+ * The delete marks the topic before releasing its leader, but a controller that read the record
+ * before the mark can win the election after the release, re-creating the lock under the record
+ * the delete is about to remove. Read the record again once elected, and give the lock back if
+ * the topic is marked deleted or gone.
+ */
+ private CompletableFuture<Void> releaseIfMarkedDeleted() {
+ return resources.getScalableTopicMetadataAsync(topicName, true)
+ .thenCompose(optMd -> {
+ if (optMd.isPresent() && !optMd.get().isDeleted()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ leaderState = LeaderElectionState.NoLeader;
+ return leaderElection.asyncClose()
+ .handle((__, ex) -> null)
+ .thenCompose(__ -> CompletableFuture.failedFuture(new IllegalStateException(
+ "Scalable topic is being deleted: " + topicName)));
+ });
+ }
+
/**
* Recover the auto split/merge/rebucket cooldown clocks after winning leadership. The
* timestamps are in-memory only, but the layout itself records when each segment was
@@ -751,8 +887,24 @@ public class ScalableTopicController {
// NamespaceService.createLookupResult(brokerId, ...), matching the pattern used by
// the cluster-leader redirection in NamespacesBase.
String brokerId = brokerService.getPulsar().getBrokerId();
+ // Registered before electing, so that no deletion is missed once the election settles.
+ resources.registerPathListener(controllerLockListener);
+ resources.registerPathListener(topicRecordListener);
+ if (closed) {
+ // close() may have deregistered them before they were registered.
+ resources.deregisterPathListener(controllerLockListener);
+ resources.deregisterPathListener(topicRecordListener);
+ return CompletableFuture.failedFuture(
+ new IllegalStateException("Controller closed for scalable topic: " + topicName));
+ }
return leaderElection.elect(brokerId)
.thenAccept(state -> {
+ if (state == LeaderElectionState.Following) {
+ // A follower holds no lock. Closing its election keeps the election from
+ // taking the lock over by itself once the lock is deleted, before this
+ // controller can check the topic: onControllerLockDeleted() re-elects.
+ leaderElection.asyncClose();
+ }
this.leaderState = state;
log.info().attr("state", state)
.log("Leader election for scalable topic");
@@ -1646,6 +1798,8 @@ public class ScalableTopicController {
public CompletableFuture<Void> close() {
closed = true;
+ resources.deregisterPathListener(controllerLockListener);
+ resources.deregisterPathListener(topicRecordListener);
cancelGcTask();
cancelAutoScaleTask();
// Stop each coordinator's drain poller before clearing — otherwise the scheduler
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
index 74b1b3859a43..76c4d1cafb5e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicService.java
@@ -113,7 +113,25 @@ public class ScalableTopicService {
CompletableFuture<ScalableTopicController> stored = controllers.computeIfAbsent(key, k -> {
ScalableTopicController controller = new ScalableTopicController(
topic, resources, brokerService, coordinationService);
- return controller.initialize().thenApply(__ -> controller);
+ CompletableFuture<ScalableTopicController> future = new CompletableFuture<>();
+ // Released once its topic is deleted: the admin delete stops the controller on the
+ // broker serving it, not the ones other brokers hold for the topic.
+ controller.setTopicDeletedAction(() -> {
+ if (controllers.remove(key, future)) {
+ controller.close();
+ }
+ });
+ controller.initialize().whenComplete((__, ex) -> {
+ if (ex == null) {
+ future.complete(controller);
+ } else {
+ // initialize() can fail after winning the election: close the controller, or
+ // its election keeps the lock.
+ controller.close();
+ future.completeExceptionally(ex);
+ }
+ });
+ return future;
});
// Evict failed futures so subsequent callers can retry. This runs *outside*
// computeIfAbsent, so modifying the map here is safe.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java
index fc3a9ce1a462..a32b52f80aac 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/SegmentLayout.java
@@ -337,8 +337,8 @@ public class SegmentLayout {
/**
* Convert back to metadata for persistence, carrying over the non-layout fields
- * (properties, per-topic auto-scale policy) from the record being replaced. Layout
- * mutations must never silently drop fields they don't model.
+ * (properties, per-topic auto-scale policy, deleted mark) from the record being replaced.
+ * Layout mutations must never silently drop fields they don't model.
*/
public ScalableTopicMetadata toMetadata(ScalableTopicMetadata original) {
return ScalableTopicMetadata.builder()
@@ -347,6 +347,7 @@ public class SegmentLayout {
.segments(new LinkedHashMap<>(allSegments))
.properties(original.getProperties())
.autoScalePolicy(original.getAutoScalePolicy())
+ .deleted(original.isDeleted())
.build();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
index 261d58cc37cd..9df2b0348567 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicControllerTest.java
@@ -48,6 +48,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.resources.ConsumerRegistration;
@@ -267,6 +268,55 @@ public class ScalableTopicControllerTest {
}
}
+ @Test
+ public void testInitializeFailsWhenTopicMarkedDeleted() throws Exception {
+ resources.markScalableTopicDeletedAsync(topicName).get();
+ try {
+ controller.initialize().get();
+ fail("expected IllegalStateException for a topic being deleted");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof IllegalStateException,
+ "expected IllegalStateException, got " + e.getCause());
+ assertTrue(e.getCause().getMessage().contains("being deleted"));
+ }
+ assertFalse(store.exists(resources.controllerLockPath(topicName)).get(),
+ "no controller may be elected for a topic being deleted");
+ }
+
+ /**
+ * A controller that read the record before the delete marked it can still win the election
+ * after the delete released the leader. It must give the lock back once elected.
+ */
+ @Test
+ public void testInitializeGivesTheLockBackWhenTopicMarkedDeletedWhileElecting() throws Exception {
+ AtomicBoolean marked = new AtomicBoolean();
+ ScalableTopicResources markingResources = new ScalableTopicResources(store, 30) {
+ @Override
+ public CompletableFuture<Optional<ScalableTopicMetadata>> getScalableTopicMetadataAsync(
+ TopicName tn, boolean refresh) {
+ // The first read sees the record unmarked, then the delete marks it.
+ return super.getScalableTopicMetadataAsync(tn, refresh)
+ .thenCompose(md -> marked.compareAndSet(false, true)
+ ? markScalableTopicDeletedAsync(tn).thenApply(__ -> md)
+ : CompletableFuture.completedFuture(md));
+ }
+ };
+ ScalableTopicController racing = new ScalableTopicController(topicName, markingResources,
+ brokerService, coordinationService);
+ try {
+ racing.initialize().get();
+ fail("expected IllegalStateException for a topic marked deleted while electing");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause().getMessage().contains("being deleted"),
+ "unexpected failure: " + e.getCause());
+ }
+ assertFalse(racing.isLeader());
+ assertFalse(store.get(resources.controllerLockPath(topicName)).get().isPresent(),
+ "the controller lock must be given back");
+ verify(scalableTopics, times(0)).createSegmentAsync(anyString(), any());
+ racing.close().join();
+ }
+
@Test
public void testLeaderBrokerIdPersisted() throws Exception {
controller.initialize().get();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
index 49b7117eb301..fbc35efe3cd8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicServiceTest.java
@@ -31,6 +31,8 @@ import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import java.time.Duration;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -59,6 +61,7 @@ import org.apache.pulsar.metadata.api.coordination.CoordinationService;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore;
+import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
@@ -364,6 +367,59 @@ public class ScalableTopicServiceTest {
verify(brokerService).deleteSchema(tn);
}
+ @Test
+ public void testControllerReleasedWhenTopicRecordDeleted() throws Exception {
+ TopicName tn = scalableTopic("t-deleted-elsewhere");
+ service.createScalableTopic(tn, 1).get();
+ service.getOrCreateController(tn).get();
+ String lockPath = resources.controllerLockPath(tn);
+ assertTrue(store.exists(lockPath).get());
+
+ // Delete the metadata the way an admin delete served by another broker does, without
+ // releasing the controller held here first.
+ resources.deleteScalableTopicAsync(tn).get();
+
+ Awaitility.await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(10))
+ .untilAsserted(() -> assertFalse(store.exists(lockPath).get(),
+ "the controller lock must not be re-acquired under the deleted topic"));
+ try {
+ service.getOrCreateController(tn).get();
+ fail("the deleted topic must not be served by its old controller");
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause().getMessage().contains("Failed to initialize controller"),
+ "got: " + e.getCause().getMessage());
+ }
+ }
+
+ /**
+ * A controller whose initialize() fails after winning the election is closed, or its election
+ * would keep the controller lock with no controller left to release it.
+ */
+ @Test
+ public void testControllerFailingAfterElectionGivesTheLockBack() throws Exception {
+ TopicName tn = scalableTopic("t-init-fails");
+ service.createScalableTopic(tn, 1).get();
+ ScalableTopicResources failingResources = new ScalableTopicResources(store, 30) {
+ @Override
+ public CompletableFuture<List<String>> listSubscriptionsAsync(TopicName topic) {
+ // Restoring the subscriptions runs once the controller won the election.
+ return CompletableFuture.failedFuture(new IllegalStateException("injected"));
+ }
+ };
+ ScalableTopicService failingService =
+ new ScalableTopicService(brokerService, failingResources, coordinationService);
+ try {
+ failingService.getOrCreateController(tn).get();
+ fail("expected the controller initialization to fail");
+ } catch (ExecutionException expected) {
+ // initialize() failed after the election
+ }
+ Awaitility.await().untilAsserted(() ->
+ assertFalse(store.get(resources.controllerLockPath(tn)).get().isPresent(),
+ "the failed controller must give the lock back"));
+ failingService.close();
+ }
+
// --- concurrent topic isolation ---
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java
index 2da9bad009cc..549152f3d3db 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/SegmentLayoutTest.java
@@ -315,6 +315,7 @@ public class SegmentLayoutTest {
// A layout mutation must round-trip every non-layout field, not just properties.
metadata.setAutoScalePolicy(org.apache.pulsar.common.policies.data.AutoScalePolicyOverride
.builder().enabled(false).maxSegments(8).build());
+ metadata.setDeleted(true);
SegmentLayout layout = SegmentLayout.fromMetadata(metadata);
SegmentLayout afterSplit = layout.splitSegment(0, 0L);
@@ -325,6 +326,7 @@ public class SegmentLayoutTest {
assertEquals(restored.getProperties().get("key"), "value");
assertEquals(restored.getAutoScalePolicy(), metadata.getAutoScalePolicy(),
"split/merge must not drop the per-topic auto-scale policy");
+ assertTrue(restored.isDeleted(), "split/merge must not drop the deleted mark");
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiBrokerScalableTopicDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiBrokerScalableTopicDeleteTest.java
new file mode 100644
index 000000000000..df59e120e54e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5MultiBrokerScalableTopicDeleteTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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.api.v5;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.resources.ScalableTopicResources;
+import org.apache.pulsar.broker.service.scalable.ScalableTopicService;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ScalableSubscriptionType;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * Deleting a scalable topic must stop its controllers before the metadata delete removes the
+ * controller lock: the leader's, also when the delete goes through another broker, and the ones
+ * following it on the other brokers, which must not take the lock over.
+ */
+public class V5MultiBrokerScalableTopicDeleteTest extends V5MultiBrokerClientBaseTest {
+
+ @Test
+ public void deleteThroughNonLeaderBrokerStopsController() throws Exception {
+ String topic = newScalableTopic(1);
+ TopicName tn = TopicName.get(topic);
+ // Elect the controller on broker 0 only: no other broker holds one for the topic.
+ ScalableTopicService leaderService = brokers.get(0).getBrokerService().getScalableTopicService();
+ leaderService.getOrCreateController(tn).get(10, TimeUnit.SECONDS);
+ ScalableTopicResources resources = brokers.get(0).getPulsarResources().getScalableTopicResources();
+ String lockPath = resources.controllerLockPath(tn);
+ assertThat(resources.getStore().exists(lockPath).get())
+ .as("controller lock after the election").isTrue();
+
+ admins.get(1).scalableTopics().deleteScalableTopic(topic, true);
+
+ Awaitility.await().during(Duration.ofSeconds(3)).atMost(Duration.ofSeconds(15))
+ .untilAsserted(() -> {
+ assertThat(resources.getStore().exists(lockPath).get())
+ .as("controller lock of the deleted topic").isFalse();
+ assertThat(resources.getStore().exists(resources.topicPath(tn)).get())
+ .as("record of the deleted topic").isFalse();
+ });
+ assertThat(admins.get(1).scalableTopics().listScalableTopics(getNamespace()))
+ .doesNotContain(topic);
+ assertThatThrownBy(() -> leaderService.getOrCreateController(tn).get(10, TimeUnit.SECONDS))
+ .hasMessageContaining("Failed to initialize controller");
+ }
+
+ /**
+ * With a controller on every broker, the delete removes the lock of the leader while the
+ * other brokers follow it. A follower's leader election takes the lock over right away, under
+ * the topic being deleted, so the record delete fails (NotEmpty on Oxia and ZooKeeper) or the
+ * lock outlives the record.
+ */
+ @Test
+ public void deleteThroughLeaderStopsFollowingControllers() throws Exception {
+ String topic = newScalableTopic(1);
+ // A subscription gives the metadata delete more to remove before it reaches the record.
+ admin.scalableTopics().createSubscription(topic, "sub", ScalableSubscriptionType.STREAM);
+ // Materializes the controller on every broker: the leader and two followers.
+ int leaderIndex = findControllerLeaderIndex(topic);
+
+ admins.get(leaderIndex).scalableTopics().deleteScalableTopic(topic, true);
+
+ assertTopicStaysDeleted(topic);
+ }
+
+ @Test
+ public void deleteThroughNonLeaderStopsFollowingControllers() throws Exception {
+ String topic = newScalableTopic(1);
+ admin.scalableTopics().createSubscription(topic, "sub", ScalableSubscriptionType.STREAM);
+ int leaderIndex = findControllerLeaderIndex(topic);
+
+ admins.get((leaderIndex + 1) % brokers.size()).scalableTopics().deleteScalableTopic(topic, true);
+
+ assertTopicStaysDeleted(topic);
+ }
+
+ /**
+ * Followers must still take over when the leader releases its controller while the topic is
+ * not being deleted (the leader broker unloads it or shuts down), and the new leader must
+ * serve as one.
+ */
+ @Test
+ public void followerTakesOverReleasedController() throws Exception {
+ String topic = newScalableTopic(1);
+ TopicName tn = TopicName.get(topic);
+ int leaderIndex = findControllerLeaderIndex(topic);
+
+ brokers.get(leaderIndex).getBrokerService().getScalableTopicService().releaseController(tn)
+ .get(10, TimeUnit.SECONDS);
+
+ ScalableTopicResources resources = brokers.get(0).getPulsarResources().getScalableTopicResources();
+ Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+ List<String> leaders = new ArrayList<>();
+ for (int i = 0; i < brokers.size(); i++) {
+ if (i != leaderIndex && brokers.get(i).getBrokerService().getScalableTopicService()
+ .getOrCreateController(tn).get(10, TimeUnit.SECONDS).isLeader()) {
+ leaders.add(brokers.get(i).getBrokerId());
+ }
+ }
+ assertThat(leaders).as("followers that took the controller over").hasSize(1);
+ Optional<GetResult> lock = resources.getStore().get(resources.controllerLockPath(tn)).get();
+ assertThat(lock).as("controller lock").isPresent();
+ assertThat(ObjectMapperFactory.getMapper().reader().readValue(lock.get().getValue(), String.class))
+ .as("broker holding the controller lock").isEqualTo(leaders.get(0));
+ });
+ }
+
+ private void assertTopicStaysDeleted(String topic) throws Exception {
+ TopicName tn = TopicName.get(topic);
+ ScalableTopicResources resources = brokers.get(0).getPulsarResources().getScalableTopicResources();
+ Awaitility.await().during(Duration.ofSeconds(3)).atMost(Duration.ofSeconds(15))
+ .untilAsserted(() -> {
+ assertThat(resources.getStore().get(resources.controllerLockPath(tn)).get())
+ .as("controller lock of the deleted topic").isEmpty();
+ assertThat(resources.getStore().get(resources.topicPath(tn)).get())
+ .as("record of the deleted topic").isEmpty();
+ });
+ assertThat(admin.scalableTopics().listScalableTopics(getNamespace())).doesNotContain(topic);
+ // No broker keeps a controller for the deleted topic: getting one creates a new controller,
+ // which fails to initialize.
+ Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+ for (PulsarService broker : brokers) {
+ assertThatThrownBy(() -> broker.getBrokerService().getScalableTopicService()
+ .getOrCreateController(tn).get(10, TimeUnit.SECONDS))
+ .as("controller of the deleted topic on " + broker.getBrokerId())
+ .hasMessageContaining("Failed to initialize controller");
+ }
+ });
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicDeleteTest.java
new file mode 100644
index 000000000000..04ed9c785382
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicDeleteTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.api.v5;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.resources.ScalableTopicResources;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.common.api.proto.ScalableConsumerAssignment;
+import org.apache.pulsar.common.api.proto.ScalableConsumerType;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.protocol.Commands;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * Deleting a scalable topic through the admin API must stop the topic's controller: once
+ * the delete returns, the controller must not re-elect itself under the deleted path, accept
+ * consumers against the deleted layout, or serve a topic re-created with the same name.
+ */
+public class V5ScalableTopicDeleteTest extends V5ClientBaseTest {
+
+ @Test
+ public void deleteReleasesElectedController() throws Exception {
+ String topic = newScalableTopic(1);
+ TopicName tn = TopicName.get(topic);
+ ScalableTopicResources resources = scalableTopicResources();
+ String lockPath = resources.controllerLockPath(tn);
+ electController(topic);
+ assertThat(resources.getStore().exists(lockPath).get())
+ .as("controller lock after the consumer registered").isTrue();
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+
+ Awaitility.await().during(Duration.ofSeconds(3)).atMost(Duration.ofSeconds(10))
+ .untilAsserted(() -> {
+ assertThat(resources.getStore().exists(lockPath).get())
+ .as("controller lock of the deleted topic").isFalse();
+ assertThat(resources.getStore().exists(resources.topicPath(tn)).get())
+ .as("record of the deleted topic").isFalse();
+ });
+ }
+
+ @Test
+ public void subscribeToDeletedTopicFails() throws Exception {
+ String topic = newScalableTopic(1);
+ TopicName tn = TopicName.get(topic);
+ electController(topic);
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+
+ assertThatThrownBy(() -> subscribe(topic, "sub", "late-consumer"))
+ .hasMessageContaining("Failed to initialize controller");
+ ScalableTopicResources resources = scalableTopicResources();
+ assertThat(resources.getStore().exists(resources.consumerPath(tn, "sub", "late-consumer")).get())
+ .as("consumer registration under the deleted topic").isFalse();
+ }
+
+ @Test
+ public void recreatedTopicGetsFreshController() throws Exception {
+ String topic = newScalableTopic(1);
+ electController(topic);
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+ admin.scalableTopics().createScalableTopic(topic, 2);
+
+ ScalableConsumerAssignment assignment = subscribe(topic, "sub", "consumer");
+ assertThat(assignment.getSegmentsCount())
+ .as("segments assigned from the re-created topic's layout").isEqualTo(2);
+ }
+
+ private ScalableTopicResources scalableTopicResources() {
+ return getPulsar().getPulsarResources().getScalableTopicResources();
+ }
+
+ /** A stream consumer registers with the topic's controller, which elects it on this broker. */
+ private void electController(String topic) throws Exception {
+ v5Client.newStreamConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("elect")
+ .subscribe()
+ .close();
+ }
+
+ /** Send a raw scalable-topic subscribe to the broker and return the assignment it replies with. */
+ private ScalableConsumerAssignment subscribe(String topic, String subscription, String consumerName)
+ throws Exception {
+ @Cleanup
+ PulsarClientImpl client = (PulsarClientImpl) newPulsarClient();
+ ClientCnx cnx = client.getConnectionToServiceUrl().get(10, TimeUnit.SECONDS);
+ long requestId = client.newRequestId();
+ return cnx.<ScalableConsumerAssignment>sendScalableSessionRequest(
+ Commands.newScalableTopicSubscribe(requestId, topic, subscription, consumerName, 1L,
+ ScalableConsumerType.STREAM),
+ requestId).get(10, TimeUnit.SECONDS);
+ }
+}