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/14 06:41:06 UTC
(pulsar) 02/14: [fix][broker] Delete the transaction state of a segment when deleting it (#26882)
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 7726b79708e86cf8160d728779b5091a3296d515
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Oct 12 22:37:58 2026 -0700
[fix][broker] Delete the transaction state of a segment when deleting it (#26882)
(cherry picked from commit 19874771ae808cc07da3411f3e45f98a26b1109a)
---
.../apache/pulsar/broker/admin/v2/Segments.java | 60 ++++
.../service/scalable/ScalableTopicService.java | 29 +-
.../buffer/impl/MetadataTransactionBuffer.java | 34 +-
.../transaction/metadata/TxnMetadataStore.java | 35 ++
.../ScalableTopicSegmentDeleteTxnStateTest.java | 364 +++++++++++++++++++++
.../buffer/impl/MetadataTransactionBufferTest.java | 140 ++++++++
.../transaction/metadata/TxnMetadataStoreTest.java | 22 ++
7 files changed, 657 insertions(+), 27 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Segments.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Segments.java
index ffc5d14664c0..6bf7a681d075 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Segments.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Segments.java
@@ -43,12 +43,16 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import lombok.CustomLog;
import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.broker.resources.ScalableTopicResources;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.metadata.TxnMetadataStore;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.common.util.FutureUtil;
/**
* Admin REST API for segment topic operations.
@@ -501,6 +505,7 @@ public class Segments extends AdminResource {
Topic t = optTopic.get();
return (force ? t.deleteForcefully() : t.delete()).thenApply(__ -> null);
})
+ .thenCompose(__ -> deleteSegmentTxnState(segmentTopic))
.thenAccept(__ -> {
log.info().attr("clientAppId", clientAppId()).attr("segment", segmentTopic)
.log("Deleted segment topic");
@@ -513,4 +518,59 @@ public class Segments extends AdminResource {
return null;
});
}
+
+ /**
+ * Delete what the transactions on a segment keep in the metadata store, which deleting the topic leaves in
+ * place: the write records of the transactions still open on it, and the state of the segment's transaction
+ * buffer. The buffer deletes a transaction's write records as it applies its outcome; left behind, they keep
+ * the coordinator from garbage-collecting the transaction, and make a segment re-created with the same name
+ * recover it as open. Only once the segment's data is gone: the aborted-txn records are what hides the data of
+ * aborted transactions.
+ *
+ * <p>Not while the segment is still active in the layout of its topic: a client can then load it again (see
+ * {@code BrokerService#isAllowAutoTopicCreationAsync}), and its new transaction buffer recovers from these
+ * records. They are left to the delete that takes the segment out of the layout. Best effort: once the topic is
+ * gone, none of the callers retries the delete.
+ */
+ private CompletableFuture<Void> deleteSegmentTxnState(TopicName segmentTopic) {
+ if (!pulsar().getConfiguration().isTransactionCoordinatorEnabled()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ String segment = segmentTopic.toString();
+ return isActiveInLayout(segmentTopic)
+ .thenCompose(active -> {
+ if (active) {
+ log.debug().attr("segment", segment)
+ .log("Keeping the transaction state of a segment still active in its topic");
+ return CompletableFuture.completedFuture(null);
+ }
+ TxnMetadataStore txnStore = new TxnMetadataStore(pulsar().getLocalMetadataStore());
+ // Independently: the write records are the ones that matter. A leftover watermark or aborted
+ // record is harmless to a later segment with the same name (ledger ids only grow, txn ids are
+ // unique).
+ return FutureUtil.waitForAll(List.of(
+ txnStore.deleteWriteOpsForSegment(segment),
+ txnStore.deleteAllSegmentState(segment)));
+ })
+ .exceptionally(ex -> {
+ log.warn().attr("segment", segment).exceptionMessage(ex)
+ .log("Failed to delete the transaction state of the deleted segment topic");
+ return null;
+ });
+ }
+
+ /** Whether the segment is active in the layout of its topic. */
+ private CompletableFuture<Boolean> isActiveInLayout(TopicName segmentTopic) {
+ ScalableTopicResources resources = pulsar().getPulsarResources().getScalableTopicResources();
+ if (resources == null) {
+ return CompletableFuture.completedFuture(false);
+ }
+ // Not from the cache: the topic delete and the controller GC update the layout from another broker, and a
+ // stale layout that still has the segment active would leave its state behind for good
+ return resources.getScalableTopicMetadataAsync(SegmentTopicName.getParentTopicName(segmentTopic), true)
+ .thenApply(optMd -> optMd
+ .map(md -> md.getSegments().get(SegmentTopicName.getSegmentId(segmentTopic)))
+ .map(seg -> seg.isActive())
+ .orElse(false));
+ }
}
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 efe523107211..786478c99f35 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
@@ -29,7 +29,6 @@ import org.apache.pulsar.broker.resources.ScalableTopicMetadata;
import org.apache.pulsar.broker.resources.ScalableTopicResources;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.GetStatsOptions;
-import org.apache.pulsar.broker.transaction.metadata.TxnMetadataStore;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.api.proto.ScalableConsumerType;
import org.apache.pulsar.common.naming.TopicDomain;
@@ -319,42 +318,26 @@ public class ScalableTopicService {
* Delete a scalable topic and all its segment topics.
*/
public CompletableFuture<Void> deleteScalableTopic(TopicName topic) {
- // When transactions are enabled, the segments carry durable /txn/segment-state records
- // (watermark + aborted-txn records). Delete them alongside the segment topics so they don't
- // outlive the data.
- TxnMetadataStore txnStore =
- brokerService.getPulsar().getConfiguration().isTransactionCoordinatorEnabled()
- ? new TxnMetadataStore(brokerService.getPulsar().getLocalMetadataStore())
- : null;
return releaseController(topic)
.thenCompose(__ -> resources.getScalableTopicMetadataAsync(topic))
+ // Delete the metadata before the segment topics, like the admin delete: the segment delete
+ // only deletes the durable transaction state of a segment no longer active in the layout.
+ .thenCompose(optMd -> resources.deleteScalableTopicAsync(topic).thenApply(__ -> optMd))
.thenCompose(optMd -> {
if (optMd.isEmpty()) {
return CompletableFuture.completedFuture(null);
}
ScalableTopicMetadata metadata = optMd.get();
- // Delete all underlying segment topics, then their durable transaction state.
+ // Delete all underlying segment topics.
return FutureUtil.waitForAll(
metadata.getSegments().values().stream()
- .map(segment -> deleteUnderlyingSegmentTopic(topic, segment)
- .thenCompose(__ -> cleanupSegmentTxnState(txnStore, topic, segment)))
+ .map(segment -> deleteUnderlyingSegmentTopic(topic, segment))
.toList()
);
})
// The segments share the schema of the scalable topic; deleting them leaves it in place.
.thenCompose(__ -> brokerService.deleteSchema(topic))
- .thenCompose(__ -> resources.deleteScalableTopicAsync(topic));
- }
-
- /** Delete the durable {@code /txn/segment-state} records for a segment being dropped. */
- private CompletableFuture<Void> cleanupSegmentTxnState(TxnMetadataStore txnStore,
- TopicName parentTopic, SegmentInfo segment) {
- if (txnStore == null) {
- return CompletableFuture.completedFuture(null);
- }
- String segmentName = SegmentTopicName.fromParent(
- parentTopic, segment.hashRange(), segment.segmentId()).toString();
- return txnStore.deleteAllSegmentState(segmentName);
+ .thenApply(__ -> null);
}
/**
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBuffer.java
index 2308c94a4f26..0aa59c9efd10 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBuffer.java
@@ -109,6 +109,11 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
private final CompletableFuture<Void> recoveryFuture = new CompletableFuture<>();
private volatile AutoCloseable subscription;
private volatile boolean closed;
+ /**
+ * Set once the segment topic is being deleted: its durable state is deleted after the topic, so the
+ * buffer stops updating it (see {@link #clearSnapshotAndClose()}).
+ */
+ private volatile boolean deleted;
/** Guards mutable state below. */
private final Object lock = new Object();
@@ -503,6 +508,12 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
// positions needed for the durable side-effects below were already captured above.
txns.remove(txnIdKey);
}
+ if (deleted) {
+ // The segment is being deleted along with its state: skip the durable side effects. Should the topic
+ // delete fail, the topic keeps serving with the outcome applied in memory, and the op records left in
+ // place let the transaction be recovered when the topic is loaded again.
+ return CompletableFuture.completedFuture(null);
+ }
// Persist aborted record if this is an abort.
CompletableFuture<Void> persistAborted = (newState == TxnState.ABORTED)
@@ -549,7 +560,7 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
long expectedVersion;
synchronized (lock) {
Position desired = maxReadPosition;
- if (desired == null) {
+ if (deleted || desired == null) {
return CompletableFuture.completedFuture(null);
}
if (watermark != null
@@ -583,10 +594,14 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
// instead of looping on the stale one. Still propagate this failure so the
// caller logs+retries; the retry now has a fresh version to work with.
return txnStore.getSegmentWatermark(segmentName).thenAccept(opt -> {
- if (opt.isPresent()) {
- synchronized (lock) {
+ synchronized (lock) {
+ if (opt.isPresent()) {
watermark = opt.get().value();
watermarkVersion = opt.get().version();
+ } else {
+ // The record was deleted (a versioned write of a missing record fails
+ // with BadVersion too): the next CAS creates it again.
+ watermarkVersion = -1L;
}
}
}).thenCompose(__ -> FutureUtil.failedFuture(cause));
@@ -806,9 +821,20 @@ public class MetadataTransactionBuffer implements TransactionBuffer {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Called when the segment topic is deleted. The durable state of the segment is not deleted here but once the
+ * segment's data is gone (see {@code Segments#deleteSegment}): the aborted-txn records are what hides the data
+ * of aborted transactions, and the topic delete can still fail. Stop updating the state and wait for the update
+ * in progress, so that none lands after the state is deleted.
+ */
@Override
public CompletableFuture<Void> clearSnapshotAndClose() {
- return closeAsync();
+ CompletableFuture<Void> updateInProgress;
+ synchronized (lock) {
+ deleted = true;
+ updateInProgress = stateTail;
+ }
+ return closeAsync().thenCompose(__ -> updateInProgress);
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStore.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStore.java
index ec7c25b389ba..6f6485e73f03 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStore.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStore.java
@@ -23,10 +23,12 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import lombok.CustomLog;
import org.apache.pulsar.common.util.FutureUtil;
@@ -187,6 +189,39 @@ public class TxnMetadataStore {
return scanAndDeleteOpsForTxn(txnId, collector -> listWritesBySegment(segment, collector));
}
+ /**
+ * Delete every {@code /txn/op} write record for {@code segment}, of any txn — used when the segment is
+ * deleted: the TB deletes them as it applies the txns' outcomes, which it never does for a txn still
+ * open by then. Best-effort: tolerates concurrent deletions via {@link MetadataStore#deleteIfExists}.
+ */
+ public CompletableFuture<Void> deleteWriteOpsForSegment(String segment) {
+ Map<String, String> txnIdByPath = new ConcurrentHashMap<>();
+ return listWritesBySegment(segment, new ScanConsumer() {
+ @Override
+ public void onNext(GetResult r) {
+ String txnIdKey = TxnPaths.txnIdFromOpPath(r.getStat().getPath());
+ if (txnIdKey != null) {
+ txnIdByPath.put(r.getStat().getPath(), txnIdKey);
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ log.warn().attr("segment", segment).exception(throwable)
+ .log("Segment write-op cleanup scan errored");
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ }).thenCompose(__ -> {
+ List<CompletableFuture<Void>> deletes = new ArrayList<>(txnIdByPath.size());
+ txnIdByPath.forEach((path, txnIdKey) -> deletes.add(
+ store.deleteIfExists(path, Optional.empty(), Set.of(new Option.PartitionKey(txnIdKey)))));
+ return FutureUtil.waitForAll(deletes);
+ });
+ }
+
/**
* Delete every {@code /txn/op} ack record for {@code (segment, subscription, txnId)} — used by
* the PendingAckStore once an event tells it the txn is terminal. Same path-extraction +
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicSegmentDeleteTxnStateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicSegmentDeleteTxnStateTest.java
new file mode 100644
index 000000000000..a1734feec620
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/scalable/ScalableTopicSegmentDeleteTxnStateTest.java
@@ -0,0 +1,364 @@
+/*
+ * 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.broker.service.scalable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import lombok.Cleanup;
+import org.apache.pulsar.broker.transaction.metadata.TxnMetadataStore;
+import org.apache.pulsar.broker.transaction.metadata.TxnPaths;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.PulsarClient;
+import org.apache.pulsar.client.api.v5.QueueConsumer;
+import org.apache.pulsar.client.api.v5.Transaction;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+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.TopicName;
+import org.apache.pulsar.common.policies.data.AutoScalePolicyOverride;
+import org.apache.pulsar.common.scalable.HashRange;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.ScanConsumer;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * Deleting the backing topic of a segment must delete the transaction state that the segment's
+ * transaction buffer keeps in the metadata store under {@link TxnPaths#TXN_SEGMENT_STATE_PREFIX}: the
+ * watermark record, written once anything is published on the segment, and one record per aborted
+ * transaction whose data is still in the segment. Nothing reads or deletes them once the segment is
+ * gone. The same goes for the {@link TxnPaths#TXN_OP_PREFIX} write records of the transactions still
+ * open on the segment, which the buffer deletes as it applies their outcome. Except while the segment
+ * is still active in the layout of its topic: a client can then load it again.
+ */
+public class ScalableTopicSegmentDeleteTxnStateTest extends V5ClientBaseTest {
+
+ @Test
+ public void testDeleteScalableTopicDeletesSegmentTxnState() throws Exception {
+ String topic = newScalableTopic(1);
+ PulsarClient client = newTxnClient();
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ String parent = abortThenSplit(topic, client, producer);
+ List<String> segments = new ArrayList<>(childrenWithTxnState(topic));
+ segments.add(parent);
+ producer.close();
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+
+ for (String segment : segments) {
+ assertThat(txnState(segment))
+ .as("transaction state of %s after deleting %s", segment, topic)
+ .isEmpty();
+ }
+ }
+
+ @Test
+ public void testDeleteSegmentDeletesTxnState() throws Exception {
+ String topic = newScalableTopic(1);
+ PulsarClient client = newTxnClient();
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ String parent = abortThenSplit(topic, client, producer);
+ List<String> children = childrenWithTxnState(topic);
+
+ admin.scalableTopics().deleteSegment(parent, true);
+
+ assertThat(txnState(parent))
+ .as("transaction state of %s after deleting its backing topic", parent)
+ .isEmpty();
+ assertChildrenKeepTxnState(children);
+ }
+
+ @Test
+ public void testDeleteActiveSegmentKeepsTxnState() throws Exception {
+ // A client can load a segment still active in the layout again right after its backing topic is deleted,
+ // and the new transaction buffer recovers from these records: they are left to the delete that takes the
+ // segment out of the layout. The re-created segment then follows the transaction still open on it.
+ String topic = newScalableTopic(1);
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ PulsarClient client = newTxnClient();
+ try (Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create()) {
+ producer.newMessage().key("k").value("before").send();
+ }
+ Awaitility.await().untilAsserted(() -> assertThat(txnState(segment))
+ .as("transaction state of %s after a publish", segment)
+ .contains("watermark"));
+ Transaction txn = writeInOpenTransaction(topic, client);
+
+ admin.scalableTopics().deleteSegment(segment, true);
+
+ assertThat(txnState(segment))
+ .as("transaction state of the active segment %s after deleting its backing topic", segment)
+ .contains("watermark");
+ assertThat(writeRecords(segment))
+ .as("write records on the active segment %s after deleting its backing topic", segment)
+ .hasSize(1);
+
+ @Cleanup
+ QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ producer.newMessage().key("k").value("after").send();
+ txn.abort();
+
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg)
+ .as("message on the re-created segment %s once the transaction open on it is aborted", segment)
+ .isNotNull();
+ assertThat(msg.value()).isEqualTo("after");
+ }
+
+ @Test
+ public void testGcPruneDeletesSegmentTxnState() throws Exception {
+ String topic = newScalableTopic(1);
+ PulsarClient client = newTxnClient();
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ String parent = abortThenSplit(topic, client, producer);
+ List<String> children = childrenWithTxnState(topic);
+
+ // The broker default retention (defaultRetentionTimeInMinutes = 0) makes the sealed parent
+ // eligible right away, and with no subscriptions there is nothing left to drain.
+ ScalableTopicController controller = getPulsar().getBrokerService().getScalableTopicService()
+ .getOrCreateController(TopicName.get(topic)).get();
+ controller.runGcTickAsync().get();
+
+ Awaitility.await().untilAsserted(() -> assertThat(getPulsar().getPulsarResources().getTopicResources()
+ .persistentTopicExists(TopicName.get(parent)).get())
+ .as("the GC tick must delete the backing topic of the sealed parent %s", parent)
+ .isFalse());
+ assertThat(txnState(parent))
+ .as("transaction state of %s after the GC tick pruned it", parent)
+ .isEmpty();
+ assertChildrenKeepTxnState(children);
+ }
+
+ @Test
+ public void testDeleteScalableTopicDeletesWriteRecordsOfOpenTransaction() throws Exception {
+ String topic = newScalableTopic(1);
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ PulsarClient client = newTxnClient();
+ Transaction txn = writeInOpenTransaction(topic, client);
+ assertThat(writeRecords(segment))
+ .as("write records on %s of a transaction still open", segment)
+ .hasSize(1);
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+
+ assertThat(writeRecords(segment))
+ .as("write records on %s of a transaction still open after deleting %s", segment, topic)
+ .isEmpty();
+ txn.abort();
+ }
+
+ @Test
+ public void testRecreatedTopicNotHeldBackByOpenTransactionOfDeletedTopic() throws Exception {
+ // A write record left behind would make the transaction buffer of the re-created segment, named like the
+ // deleted one, recover the transaction as open, and hold back every message until the transaction ends.
+ String topic = newScalableTopic(1);
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ PulsarClient client = newTxnClient();
+ Transaction txn = writeInOpenTransaction(topic, client);
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+ admin.scalableTopics().createScalableTopic(topic, 1);
+ assertThat(segmentTopicName(topic, singleActiveSegmentId(topic)))
+ .as("segment of %s re-created with the same name", topic)
+ .isEqualTo(segment);
+
+ @Cleanup
+ QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ producer.newMessage().key("k").value("v").send();
+
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg)
+ .as("message on %s, re-created while a transaction that wrote to it is still open", topic)
+ .isNotNull();
+ assertThat(msg.value()).isEqualTo("v");
+ txn.abort();
+ }
+
+ private PulsarClient newTxnClient() throws Exception {
+ return track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .transactionPolicy(TransactionPolicy.builder().timeout(Duration.ofMinutes(2)).build())
+ .build());
+ }
+
+ /**
+ * Aborts a transaction that wrote to the single initial segment, waits until the segment has
+ * persisted its watermark and the aborted-transaction record, then splits the segment and
+ * publishes on the children. Returns the name of the sealed parent's backing topic.
+ */
+ private String abortThenSplit(String topic, PulsarClient client, Producer<String> producer)
+ throws Exception {
+ admin.scalableTopics().setAutoScalePolicy(topic,
+ AutoScalePolicyOverride.builder().enabled(false).build());
+ long parentId = singleActiveSegmentId(topic);
+ String parent = segmentTopicName(topic, parentId);
+
+ Transaction txn = client.newTransaction();
+ producer.newMessage().key("k").transaction(txn).value("aborted").send();
+ txn.abort();
+ Awaitility.await().untilAsserted(() -> assertThat(txnState(parent))
+ .as("transaction state of %s after aborting a transaction that wrote to it", parent)
+ .contains("watermark")
+ .anyMatch(record -> record.startsWith("aborted:")));
+
+ admin.scalableTopics().splitSegment(topic, parentId);
+ Awaitility.await().untilAsserted(() -> assertThat(activeSegmentIds(topic))
+ .as("active segments after splitting segment %s", parentId)
+ .hasSize(2));
+ producer.newMessage().key("k").value("post").send();
+ return parent;
+ }
+
+ /** Publishes a message on {@code topic} in a transaction left open, and returns the transaction. */
+ private Transaction writeInOpenTransaction(String topic, PulsarClient client) throws Exception {
+ Transaction txn = client.newTransaction();
+ try (Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create()) {
+ producer.newMessage().key("k").transaction(txn).value("open").send();
+ }
+ return txn;
+ }
+
+ /** The active segments that have persisted a watermark for what was published on them. */
+ private List<String> childrenWithTxnState(String topic) throws Exception {
+ List<String> children = new ArrayList<>();
+ for (long id : activeSegmentIds(topic)) {
+ children.add(segmentTopicName(topic, id));
+ }
+ List<String> withState = new ArrayList<>();
+ Awaitility.await().untilAsserted(() -> {
+ withState.clear();
+ for (String child : children) {
+ if (!txnState(child).isEmpty()) {
+ withState.add(child);
+ }
+ }
+ assertThat(withState).as("children of the split with transaction state").isNotEmpty();
+ });
+ return withState;
+ }
+
+ private void assertChildrenKeepTxnState(List<String> children) throws Exception {
+ for (String child : children) {
+ assertThat(txnState(child))
+ .as("transaction state of the active segment %s after deleting its sealed parent", child)
+ .contains("watermark");
+ }
+ }
+
+ /**
+ * The records the segment's transaction buffer keeps under {@link TxnPaths#TXN_SEGMENT_STATE_PREFIX}:
+ * {@code "watermark"} for the watermark record and {@code "aborted:<txnId>"} for each aborted
+ * transaction.
+ */
+ private List<String> txnState(String segment) throws Exception {
+ TxnMetadataStore txnStore = new TxnMetadataStore(getPulsar().getLocalMetadataStore());
+ List<String> records = Collections.synchronizedList(new ArrayList<>());
+ if (txnStore.getSegmentWatermark(segment).get().isPresent()) {
+ records.add("watermark");
+ }
+ txnStore.scanAbortedTxns(segment,
+ TxnPaths.abortedByPositionSegmentLowerBound(segment),
+ TxnPaths.abortedByPositionSegmentUpperBound(segment),
+ new ScanConsumer() {
+ @Override
+ public void onNext(GetResult r) {
+ records.add("aborted:" + TxnPaths.txnIdFromAbortedPath(r.getStat().getPath()));
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ }).get();
+ return records;
+ }
+
+ /** The paths of the {@link TxnPaths#TXN_OP_PREFIX} write records on the segment. */
+ private List<String> writeRecords(String segment) throws Exception {
+ TxnMetadataStore txnStore = new TxnMetadataStore(getPulsar().getLocalMetadataStore());
+ List<String> paths = Collections.synchronizedList(new ArrayList<>());
+ txnStore.listWritesBySegment(segment, new ScanConsumer() {
+ @Override
+ public void onNext(GetResult r) {
+ paths.add(r.getStat().getPath());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ }).get();
+ return paths;
+ }
+
+ private long singleActiveSegmentId(String topic) throws Exception {
+ List<Long> active = activeSegmentIds(topic);
+ assertThat(active).as("active segments of %s", topic).hasSize(1);
+ return active.get(0);
+ }
+
+ private List<Long> activeSegmentIds(String topic) throws Exception {
+ return admin.scalableTopics().getMetadata(topic).getSegments().values().stream()
+ .filter(seg -> seg.isActive())
+ .map(seg -> seg.getSegmentId())
+ .toList();
+ }
+
+ private String segmentTopicName(String topic, long segmentId) throws Exception {
+ var range = admin.scalableTopics().getMetadata(topic).getSegments().get(segmentId).getHashRange();
+ return SegmentTopicName.fromParent(TopicName.get(topic),
+ HashRange.of(range.getStart(), range.getEnd()), segmentId).toString();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBufferTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBufferTest.java
index a7f3890111b2..65ec32f45855 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBufferTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/MetadataTransactionBufferTest.java
@@ -21,13 +21,23 @@ package org.apache.pulsar.broker.transaction.buffer.impl;
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.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
@@ -38,6 +48,7 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.transaction.metadata.AbortedTxnRecord;
+import org.apache.pulsar.broker.transaction.metadata.SegmentWatermark;
import org.apache.pulsar.broker.transaction.metadata.TxnEvent;
import org.apache.pulsar.broker.transaction.metadata.TxnHeader;
import org.apache.pulsar.broker.transaction.metadata.TxnIds;
@@ -47,9 +58,11 @@ import org.apache.pulsar.broker.transaction.metadata.TxnOpKind;
import org.apache.pulsar.broker.transaction.metadata.TxnPaths;
import org.apache.pulsar.broker.transaction.metadata.TxnState;
import org.apache.pulsar.client.api.transaction.TxnID;
+import org.apache.pulsar.metadata.api.GetResult;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
+import org.apache.pulsar.metadata.api.ScanConsumer;
import org.apache.pulsar.metadata.api.Stat;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterMethod;
@@ -309,6 +322,126 @@ public class MetadataTransactionBufferTest {
assertThat(tb2.isTxnAborted(liveTxn, PositionFactory.create(10, 5))).isTrue();
}
+ @Test
+ public void clearSnapshotAndClose_waitsForStateUpdateInProgress_thenStopsUpdates() throws Exception {
+ // The segment delete deletes the durable state after the topic: deleting the topic must wait for the
+ // state update in progress, and no update may start afterwards, or it would re-create the state.
+ TxnMetadataStore spyStore = spy(txnStore);
+ CompletableFuture<Stat> watermarkWrite = new CompletableFuture<>();
+ doReturn(watermarkWrite).when(spyStore).casSegmentWatermark(eq(SEGMENT), any(), any());
+ MetadataTransactionBuffer tb = new MetadataTransactionBuffer(topic, spyStore);
+ tb.checkIfTBRecoverCompletely().get();
+
+ when(ledger.getLastConfirmedEntry()).thenReturn(PositionFactory.create(10, 1));
+ tb.syncMaxReadPositionForNormalPublish(PositionFactory.create(10, 1), false);
+ verify(spyStore).casSegmentWatermark(eq(SEGMENT), any(), any());
+
+ CompletableFuture<Void> deleted = tb.clearSnapshotAndClose();
+ assertThat(deleted).as("topic delete with a watermark write in progress").isNotDone();
+ watermarkWrite.complete(new Stat(TxnPaths.segmentWatermarkPath(SEGMENT), 0, 0, 0, false, true));
+ deleted.get();
+
+ // No update starts once the topic delete has started
+ when(ledger.getLastConfirmedEntry()).thenReturn(PositionFactory.create(10, 2));
+ tb.syncMaxReadPositionForNormalPublish(PositionFactory.create(10, 2), false);
+ verify(spyStore, times(1)).casSegmentWatermark(eq(SEGMENT), any(), any());
+ }
+
+ @Test
+ public void clearSnapshotAndClose_leavesStateOfTxnRecoveredAfterwards() throws Exception {
+ // A terminal txn that recovery finds only once the topic delete has started: its outcome is neither
+ // persisted nor are its op records deleted, so that it is recovered when the topic is loaded again, should
+ // the delete fail.
+ TxnID txnId = new TxnID(1, 1);
+ createOpenHeader(txnId);
+ Stat opRecord = txnStore.appendOp(TxnIds.toKey(txnId),
+ new TxnOp(TxnOpKind.WRITE, SEGMENT, null, 0L, 0L, null)).get();
+ abortTxn(txnId);
+ TxnMetadataStore spyStore = spy(txnStore);
+ CompletableFuture<Void> opScanReleased = new CompletableFuture<>();
+ doAnswer(inv -> opScanReleased.thenCompose(__ ->
+ txnStore.listWritesBySegment(SEGMENT, inv.getArgument(1))))
+ .when(spyStore).listWritesBySegment(eq(SEGMENT), any());
+ MetadataTransactionBuffer tb = new MetadataTransactionBuffer(topic, spyStore);
+
+ tb.clearSnapshotAndClose().get();
+ opScanReleased.complete(null);
+ // Recovery applies the terminal txns it finds before it completes: stateTail was complete
+ tb.checkIfTBRecoverCompletely().get();
+
+ verify(spyStore, never()).putAbortedTxn(any(), any(), anyLong(), anyLong());
+ assertThat(store.exists(opRecord.getPath()).get()).as("op record of the recovered txn").isTrue();
+ }
+
+ @Test
+ public void clearSnapshotAndClose_stillAppliesQueuedOutcomeInMemory() throws Exception {
+ // An outcome queued behind the state update in progress when the topic delete starts. Should the delete
+ // fail, the topic keeps serving with this buffer: the outcome must still apply in memory, or the txn stays
+ // open and keeps pinning maxReadPosition. Only its durable side effects are skipped.
+ TxnID txnId = new TxnID(1, 1);
+ createOpenHeader(txnId);
+ TxnMetadataStore spyStore = spy(txnStore);
+ CompletableFuture<Stat> watermarkWrite = new CompletableFuture<>();
+ doReturn(watermarkWrite).when(spyStore).casSegmentWatermark(eq(SEGMENT), any(), any());
+ MetadataTransactionBuffer tb = new MetadataTransactionBuffer(topic, spyStore);
+ tb.checkIfTBRecoverCompletely().get();
+ Position txnPos = tb.appendBufferToTxn(txnId, 0, payload("a")).get();
+ when(ledger.getLastConfirmedEntry()).thenReturn(PositionFactory.create(10, 2));
+ tb.syncMaxReadPositionForNormalPublish(PositionFactory.create(10, 2), false);
+ verify(spyStore).casSegmentWatermark(eq(SEGMENT), any(), any());
+
+ abortTxn(txnId);
+ txnStore.publishSegmentEvent(SEGMENT, new TxnEvent(TxnIds.toKey(txnId), TxnState.ABORTED)).get();
+ // The reconcile triggered by the event reads the header (the first read is the append's) and queues the
+ // outcome behind the watermark write
+ verify(spyStore, timeout(10_000).atLeast(2)).getHeader(TxnIds.toKey(txnId));
+ CompletableFuture<Void> deleted = tb.clearSnapshotAndClose();
+ watermarkWrite.complete(new Stat(TxnPaths.segmentWatermarkPath(SEGMENT), 0, 0, 0, false, true));
+ deleted.get();
+
+ Awaitility.await().untilAsserted(() -> {
+ assertThat(tb.getOngoingTxnCount()).as("open txns once the abort is applied").isZero();
+ assertThat(tb.isTxnAborted(txnId, txnPos)).isTrue();
+ });
+ assertThat(tb.getMaxReadPosition()).isEqualTo(PositionFactory.create(10, 2));
+ verify(spyStore, never()).putAbortedTxn(any(), any(), anyLong(), anyLong());
+ List<String> writeRecords = new ArrayList<>();
+ txnStore.listWritesBySegment(SEGMENT, new ScanConsumer() {
+ @Override
+ public void onNext(GetResult r) {
+ writeRecords.add(r.getStat().getPath());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ }).get();
+ assertThat(writeRecords).as("op records of the aborted txn").hasSize(1);
+ }
+
+ @Test
+ public void watermarkRecordDeletedUnderLiveBuffer_isCreatedAgain() throws Exception {
+ // A versioned write of a record that is gone fails with BadVersion, like one against a newer version: the
+ // buffer must then create the record again rather than keep writing against the version of the deleted one.
+ MetadataTransactionBuffer tb = new MetadataTransactionBuffer(topic, txnStore);
+ tb.checkIfTBRecoverCompletely().get();
+ publish(tb, 1);
+ Awaitility.await().untilAsserted(() -> assertThat(txnStore.getSegmentWatermark(SEGMENT).get()).isPresent());
+ txnStore.deleteSegmentWatermark(SEGMENT).get();
+
+ // The write after the delete finds the record gone, the next one creates it again
+ publish(tb, 2);
+ publish(tb, 3);
+
+ Awaitility.await().untilAsserted(() -> assertThat(txnStore.getSegmentWatermark(SEGMENT).get())
+ .as("watermark record after it was deleted under the live buffer")
+ .hasValueSatisfying(v -> assertThat(v.value()).isEqualTo(new SegmentWatermark(10, 3))));
+ }
+
@Test
public void recoveryDiscoveredOpenTxn_pinsAtWatermark() throws Exception {
// /txn/op records exist for an open txn (broker was processing publishes for txn T;
@@ -462,6 +595,13 @@ public class MetadataTransactionBufferTest {
return Unpooled.copiedBuffer(s.getBytes());
}
+ /** A non-transactional publish at {@code 10:entryId}, the last entry of the segment from then on. */
+ private void publish(MetadataTransactionBuffer tb, long entryId) {
+ Position position = PositionFactory.create(10, entryId);
+ when(ledger.getLastConfirmedEntry()).thenReturn(position);
+ tb.syncMaxReadPositionForNormalPublish(position, false);
+ }
+
private void createOpenHeader(TxnID txnId) throws Exception {
txnStore.createHeader(TxnIds.toKey(txnId),
new TxnHeader(TxnState.OPEN, Duration.ofMillis(60_000),
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStoreTest.java
index 19d22f3f46e9..38d0bbd36f7b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnMetadataStoreTest.java
@@ -112,6 +112,28 @@ public class TxnMetadataStoreTest {
assertThat(hits).containsExactlyInAnyOrder(w1, w2);
}
+ @Test
+ public void deleteWriteOpsForSegment_removesTheSegmentsWriteOpsOfEveryTxn() throws Exception {
+ @Cleanup MetadataStore store = newMemoryStore();
+ TxnMetadataStore txn = new TxnMetadataStore(store);
+
+ String segA = "segment://public/default/topic/0000-7fff-0";
+ String segB = "segment://public/default/topic/8000-ffff-0";
+ txn.appendOp("tx-1", new TxnOp(TxnOpKind.WRITE, segA, null, 0L, 0L, null)).get();
+ txn.appendOp("tx-2", new TxnOp(TxnOpKind.WRITE, segA, null, 0L, 0L, null)).get();
+ TxnOp otherSegmentWrite = new TxnOp(TxnOpKind.WRITE, segB, null, 0L, 0L, null);
+ txn.appendOp("tx-1", otherSegmentWrite).get();
+ TxnOp ack = new TxnOp(TxnOpKind.ACK, segA, "sub/x", 1L, 5L, false);
+ txn.appendOp("tx-1", ack).get();
+
+ txn.deleteWriteOpsForSegment(segA).get();
+
+ List<TxnOp> remaining = new ArrayList<>();
+ txn.listOpsByTxn("tx-1", collectOps(remaining)).get();
+ txn.listOpsByTxn("tx-2", collectOps(remaining)).get();
+ assertThat(remaining).containsExactlyInAnyOrder(otherSegmentWrite, ack);
+ }
+
@Test
public void appendAndScanAckOps() throws Exception {
@Cleanup MetadataStore store = newMemoryStore();