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:11 UTC
(pulsar) 07/14: [fix][broker] Delete the ack records of open transactions when deleting a segment or a subscription (#26887)
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 228c3945b3a80c6e31097c15e61ea6d35d4b3729
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Oct 13 14:34:07 2026 -0700
[fix][broker] Delete the ack records of open transactions when deleting a segment or a subscription (#26887)
(cherry picked from commit ebcf801a7f7f3a85b94e1367aeddee6aab2704d6)
---
.../apache/pulsar/broker/admin/v2/Segments.java | 18 +--
.../service/persistent/PersistentSubscription.java | 39 +++++-
.../transaction/metadata/TxnMetadataStore.java | 134 +++++++++++++++----
.../broker/transaction/metadata/TxnPaths.java | 18 +++
.../pendingack/impl/MetadataPendingAckStore.java | 39 +++++-
...SegmentSubscriptionDeleteTxnAckRecordsTest.java | 145 +++++++++++++++++++++
.../ScalableTopicSegmentDeleteTxnStateTest.java | 84 +++++++++++-
.../transaction/metadata/TxnMetadataStoreTest.java | 118 +++++++++++++++++
.../broker/transaction/metadata/TxnPathsTest.java | 44 +++++++
.../impl/MetadataPendingAckStoreTest.java | 48 +++++++
10 files changed, 644 insertions(+), 43 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 6bf7a681d075..24b4402fc6c5 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
@@ -521,11 +521,12 @@ public class Segments extends AdminResource {
/**
* 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.
+ * place: the write records of the transactions still open on it, their ack records on its subscriptions, and
+ * the state of the segment's transaction buffer. The buffer and the pending-ack store of each subscription
+ * delete a transaction's records as they apply 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
@@ -545,11 +546,12 @@ public class Segments extends AdminResource {
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).
+ // Independently: the write and ack 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.deleteAckOpsForSegment(segment),
txnStore.deleteAllSegmentState(segment)));
})
.exceptionally(ex -> {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 0f3135d4c0f2..75b9dba4c116 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -74,6 +74,7 @@ import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.plugin.EntryFilter;
import org.apache.pulsar.broker.stats.OpenTelemetryMessageFinderStats.FindReason;
+import org.apache.pulsar.broker.transaction.metadata.TxnMetadataStore;
import org.apache.pulsar.broker.transaction.pendingack.PendingAckHandle;
import org.apache.pulsar.broker.transaction.pendingack.impl.PendingAckHandleDisabled;
import org.apache.pulsar.broker.transaction.pendingack.impl.PendingAckHandleImpl;
@@ -136,6 +137,8 @@ public class PersistentSubscription extends AbstractSubscription {
private volatile CompletableFuture<Void> inProgressResetCursorFuture;
private volatile Boolean replicatedControlled;
private final ServiceConfiguration config;
+ /** Held by tests between the cursor delete and the delete of the txn ack records, see {@link #unsubscribe}. */
+ private volatile CompletableFuture<Void> txnAckRecordsDeleteGate = CompletableFuture.completedFuture(null);
static Map<String, Long> getBaseCursorProperties(Boolean isReplicated) {
return isReplicated != null && isReplicated ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES :
@@ -1291,7 +1294,7 @@ public class PersistentSubscription extends AbstractSubscription {
}
// cursor close handles pending delete (ack) operations
- closeSubscriptionFuture.thenCompose(v -> topic.unsubscribe(subName)).thenAccept(v -> {
+ closeSubscriptionFuture.thenCompose(v -> unsubscribe()).thenAccept(v -> {
synchronized (this) {
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null)).thenRun(() -> {
log.info()
@@ -1321,6 +1324,40 @@ public class PersistentSubscription extends AbstractSubscription {
return deleteFuture;
}
+ /**
+ * Delete the cursor of the subscription, once closed, and with it the ack records that the pending-ack store of
+ * a segment's subscription keeps in the metadata store for the transactions still open: the store deletes a
+ * transaction's records as it applies its outcome, which it never does once the subscription is gone. The records
+ * are collected first, while the subscription is fenced and its store closed: a subscription re-created with the
+ * same name once the cursor is deleted writes records of its own. They are deleted once the cursor is, so that a
+ * subscription whose delete fails keeps them. Best effort: a failure is logged, not returned.
+ */
+ private CompletableFuture<Void> unsubscribe() {
+ if (!config.isTransactionCoordinatorEnabled() || !TopicName.get(topicName).isSegment()) {
+ return topic.unsubscribe(subName);
+ }
+ TxnMetadataStore txnStore = new TxnMetadataStore(topic.getBrokerService().getPulsar().getLocalMetadataStore());
+ return txnStore.collectAckOpsForSegmentSubscription(topicName, subName)
+ .exceptionally(ex -> {
+ log.warn().exceptionMessage(ex)
+ .log("Failed to collect the transaction ack records of the subscription to delete");
+ return List.of();
+ })
+ .thenCompose(ackOps -> topic.unsubscribe(subName)
+ .thenCompose(__ -> txnAckRecordsDeleteGate)
+ .thenCompose(__ -> txnStore.deleteOps(ackOps)
+ .exceptionally(ex -> {
+ log.warn().exceptionMessage(ex)
+ .log("Failed to delete the transaction ack records of the subscription");
+ return null;
+ })));
+ }
+
+ @VisibleForTesting
+ void setTxnAckRecordsDeleteGate(CompletableFuture<Void> gate) {
+ this.txnAckRecordsDeleteGate = gate;
+ }
+
/**
* Handle unsubscribe command from the client API Check with the dispatcher is this consumer can proceed with
* unsubscribe.
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 6f6485e73f03..6233148ba7ee 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
@@ -30,6 +30,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
import lombok.CustomLog;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -65,6 +67,13 @@ public class TxnMetadataStore {
private final MetadataStore store;
+ /**
+ * A {@code /txn/op} record collected to be deleted: its path, the txn it belongs to, and the version it was read
+ * at. The path alone doesn't identify the record: Oxia gives the next op of a txn the sequence key of the txn's
+ * last op once that one is deleted.
+ */
+ public record OpRecord(String path, String txnId, long version) {}
+
public TxnMetadataStore(MetadataStore store) {
this.store = store;
}
@@ -192,34 +201,13 @@ public class TxnMetadataStore {
/**
* 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}.
+ * open by then. Best-effort, see {@link #deleteOps}.
*/
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);
- });
+ return collectOpsOnSegment(segment,
+ op -> op.getKind() == TxnOpKind.WRITE && segment.equals(op.getSegment()),
+ collector -> listWritesBySegment(segment, collector))
+ .thenCompose(this::deleteOps);
}
/**
@@ -233,6 +221,64 @@ public class TxnMetadataStore {
collector -> listAcksBySegmentSubscription(segment, subscription, collector));
}
+ /**
+ * Collect every {@code /txn/op} ack record for {@code (segment, subscription)}, of any txn, for
+ * {@link #deleteOps} — used when the subscription is deleted: the PendingAckStore deletes them as it applies the
+ * txns' outcomes, which it never does for a txn still open by then. A snapshot: the records a subscription
+ * re-created with the same name writes afterwards aren't in it.
+ */
+ public CompletableFuture<List<OpRecord>> collectAckOpsForSegmentSubscription(String segment,
+ String subscription) {
+ return collectOpsOnSegment(segment,
+ op -> op.getKind() == TxnOpKind.ACK && segment.equals(op.getSegment())
+ && subscription.equals(op.getSubscription()),
+ collector -> listAcksBySegmentSubscription(segment, subscription, collector));
+ }
+
+ /**
+ * Delete every {@code /txn/op} ack record for {@code segment}, of any subscription and txn — used when the
+ * segment is deleted, for the same reason as {@link #collectAckOpsForSegmentSubscription}. A single scan,
+ * whatever the number of subscriptions, which also finds the records of subscriptions deleted before.
+ */
+ public CompletableFuture<Void> deleteAckOpsForSegment(String segment) {
+ return collectOpsOnSegment(segment,
+ op -> op.getKind() == TxnOpKind.ACK && segment.equals(op.getSegment()),
+ collector -> listAcksBySegment(segment, collector))
+ .thenCompose(this::deleteOps);
+ }
+
+ /**
+ * Delete the {@code /txn/op} records collected by {@link #collectAckOpsForSegmentSubscription}, each at the
+ * version it was collected at. Best-effort: a record deleted since is left alone, and so is a record written at
+ * its path since, for another op of the same txn.
+ */
+ public CompletableFuture<Void> deleteOps(List<OpRecord> ops) {
+ List<CompletableFuture<Void>> deletes = new ArrayList<>(ops.size());
+ for (OpRecord op : ops) {
+ deletes.add(store.delete(op.path(), Optional.of(op.version()), Set.of(new Option.PartitionKey(op.txnId())))
+ .exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ if (cause instanceof MetadataStoreException.NotFoundException
+ || cause instanceof MetadataStoreException.BadVersionException) {
+ return null;
+ }
+ throw FutureUtil.wrapToCompletionException(cause);
+ }));
+ }
+ return FutureUtil.waitForAll(deletes);
+ }
+
+ /** Stream all ack ops targeting {@code segment}, of any subscription. */
+ private CompletableFuture<Void> listAcksBySegment(String segment, ScanConsumer consumer) {
+ return store.scanByIndex(TxnPaths.TXN_OP_PREFIX, TxnPaths.IDX_ACKS_BY_SEGMENT_SUBSCRIPTION,
+ TxnPaths.ackIndexSegmentLowerBound(segment), TxnPaths.ackIndexSegmentUpperBound(segment),
+ gr -> {
+ TxnOp op = fromJson(gr.getValue(), TxnOp.class);
+ return op.getKind() == TxnOpKind.ACK && segment.equals(op.getSegment());
+ },
+ consumer);
+ }
+
/**
* Shared implementation: invoke the supplied scan with a collector that captures only paths
* matching {@code txnId}, then delete each captured path with the txn-scoped partition key.
@@ -271,6 +317,42 @@ public class TxnMetadataStore {
});
}
+ /**
+ * Shared implementation of the cleanups on a deleted segment or subscription, of any txn: invoke the supplied
+ * scan with a collector that captures the records {@code matches} accepts. The collector checks the records
+ * itself: a store with a native index doesn't apply the scan's filter, and its point lookups also match the keys
+ * that extend the one looked up (the records of a segment whose id extends this one's, of a subscription whose
+ * name extends this one's).
+ */
+ private CompletableFuture<List<OpRecord>> collectOpsOnSegment(
+ String segment, Predicate<TxnOp> matches, Function<ScanConsumer, CompletableFuture<Void>> scan) {
+ Map<String, OpRecord> opsByPath = new ConcurrentHashMap<>();
+ ScanConsumer collector = new ScanConsumer() {
+ @Override
+ public void onNext(GetResult r) {
+ if (!matches.test(fromJson(r.getValue(), TxnOp.class))) {
+ return;
+ }
+ String path = r.getStat().getPath();
+ String txnIdKey = TxnPaths.txnIdFromOpPath(path);
+ if (txnIdKey != null) {
+ opsByPath.put(path, new OpRecord(path, txnIdKey, r.getStat().getVersion()));
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ log.warn().attr("segment", segment).exception(throwable)
+ .log("Segment op-record cleanup scan errored");
+ }
+
+ @Override
+ public void onCompleted() {
+ }
+ };
+ return scan.apply(collector).thenApply(__ -> List.copyOf(opsByPath.values()));
+ }
+
/**
* Stream open transactions whose deadline falls in {@code [fromMsInclusive, toMsInclusive]}.
* Pass {@code null} on either bound for an unbounded range.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java
index e94d0b170539..cc869904d728 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java
@@ -198,6 +198,24 @@ public final class TxnPaths {
return segmentKey(segment) + ":" + Codec.encode(subscription);
}
+ /**
+ * @return the lower bound of the {@link #IDX_ACKS_BY_SEGMENT_SUBSCRIPTION} keys of {@code segment}, of any
+ * subscription. The key of a segment whose id extends this one's continues with a digit, which sorts before
+ * the {@code :} separator.
+ */
+ public static String ackIndexSegmentLowerBound(String segment) {
+ return segmentKey(segment) + ":";
+ }
+
+ /**
+ * @return the upper bound of the {@link #IDX_ACKS_BY_SEGMENT_SUBSCRIPTION} keys of {@code segment}, of any
+ * subscription. URL encoding escapes {@code ~}, which sorts after every character it leaves in a
+ * subscription name.
+ */
+ public static String ackIndexSegmentUpperBound(String segment) {
+ return segmentKey(segment) + ":~";
+ }
+
/** Parent path for per-segment watermark records. Records are direct children of this. */
public static final String TXN_SEGMENT_WATERMARK_PREFIX = TXN_SEGMENT_STATE_PREFIX + "/watermark";
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStore.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStore.java
index 439b75ecab23..1f8521ce4bd6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStore.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStore.java
@@ -83,6 +83,12 @@ public class MetadataPendingAckStore implements PendingAckStore {
private final Object lock = new Object();
/** Set of txnIdKeys we believe are OPEN — populated on append + recovery, drained by reconcile. */
private final Set<String> openTxns = new HashSet<>();
+ /**
+ * The ack appends in progress, which {@link #closeAsync()} waits for: the ack records of a deleted subscription
+ * are deleted once it is closed, and none must land after that. Guarded by {@link #lock}, like the
+ * {@link #closed} check of an append.
+ */
+ private final Set<CompletableFuture<Void>> appendsInProgress = new HashSet<>();
private final CompletableFuture<Void> recoveryFuture = new CompletableFuture<>();
private volatile AutoCloseable eventSubscription;
@@ -201,8 +207,12 @@ public class MetadataPendingAckStore implements PendingAckStore {
private CompletableFuture<Void> appendAcks(TxnID txnID, List<MutablePair<Position, Integer>> positions,
boolean cumulative) {
- if (closed) {
- return FutureUtil.failedFuture(new IllegalStateException("PendingAckStore is closed"));
+ CompletableFuture<Void> appended = new CompletableFuture<>();
+ synchronized (lock) {
+ if (closed) {
+ return FutureUtil.failedFuture(new IllegalStateException("PendingAckStore is closed"));
+ }
+ appendsInProgress.add(appended);
}
String txnIdKey = TxnIds.toKey(txnID);
// Append one TxnOp per position. PendingAckHandleImpl already validates the txn isn't
@@ -214,11 +224,20 @@ public class MetadataPendingAckStore implements PendingAckStore {
p.getLedgerId(), p.getEntryId(), cumulative ? Boolean.TRUE : null);
appends[i] = txnStore.appendOp(txnIdKey, op);
}
- return CompletableFuture.allOf(appends).thenRun(() -> {
+ CompletableFuture.allOf(appends).whenComplete((__, err) -> {
synchronized (lock) {
- openTxns.add(txnIdKey);
+ appendsInProgress.remove(appended);
+ if (err == null) {
+ openTxns.add(txnIdKey);
+ }
+ }
+ if (err != null) {
+ appended.completeExceptionally(err);
+ } else {
+ appended.complete(null);
}
});
+ return appended;
}
@Override
@@ -310,11 +329,19 @@ public class MetadataPendingAckStore implements PendingAckStore {
// ---- Close -------------------------------------------------------------
+ /**
+ * Completes once the ack appends in progress are done; no append starts after it is called. A failed append
+ * fails the ack it was for, not the close.
+ */
@Override
public CompletableFuture<Void> closeAsync() {
- closed = true;
+ CompletableFuture<?>[] inProgress;
+ synchronized (lock) {
+ closed = true;
+ inProgress = appendsInProgress.toArray(new CompletableFuture<?>[0]);
+ }
closeSubscriptionQuietly();
- return CompletableFuture.completedFuture(null);
+ return CompletableFuture.allOf(inProgress).exceptionally(__ -> null);
}
private void closeSubscriptionQuietly() {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SegmentSubscriptionDeleteTxnAckRecordsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SegmentSubscriptionDeleteTxnAckRecordsTest.java
new file mode 100644
index 000000000000..09fe5daa7505
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/SegmentSubscriptionDeleteTxnAckRecordsTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.persistent;
+
+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 java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+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.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 a subscription of a segment deletes the {@link TxnPaths#TXN_OP_PREFIX} ack records of the transactions
+ * still open on it, and only those: the cursor delete lets a subscription with the same name be created again, and
+ * the records of the new subscription must stay.
+ */
+public class SegmentSubscriptionDeleteTxnAckRecordsTest extends V5ClientBaseTest {
+
+ @Test
+ public void testDeleteSubscriptionKeepsAckRecordsOfSubscriptionRecreatedWithSameName() throws Exception {
+ String topic = newScalableTopic(1);
+ String segment = singleSegment(topic);
+ PulsarClient client = track(PulsarClient.builder()
+ .serviceUrl(getBrokerServiceUrl())
+ .transactionPolicy(TransactionPolicy.builder().timeout(Duration.ofMinutes(2)).build())
+ .build());
+ Transaction oldTxn = client.newTransaction();
+ ackInTransaction(topic, client, "sub", oldTxn);
+ List<String> oldRecords = ackRecords(segment, "sub");
+ assertThat(oldRecords).as("ack records of the subscription to delete").hasSize(1);
+
+ // Hold the delete once the cursor is deleted, before the ack records are.
+ PersistentTopic segmentTopic = (PersistentTopic) getTopicReference(segment).orElseThrow();
+ CompletableFuture<Void> gate = new CompletableFuture<>();
+ segmentTopic.getSubscription("sub").setTxnAckRecordsDeleteGate(gate);
+ CompletableFuture<Void> delete = admin.scalableTopics().deleteSubscriptionAsync(topic, "sub");
+ Awaitility.await().untilAsserted(() -> assertThat(segmentTopic.getSubscription("sub"))
+ .as("subscription of %s once its cursor is deleted", segment)
+ .isNull());
+
+ // A subscription created again with the same name acknowledges in a transaction of its own.
+ Transaction newTxn = client.newTransaction();
+ ackInTransaction(topic, client, "sub", newTxn);
+ gate.complete(null);
+ delete.get(30, TimeUnit.SECONDS);
+
+ assertThat(ackRecords(segment, "sub"))
+ .as("ack records on %s of the subscription created again with the same name", segment)
+ .hasSize(1)
+ .doesNotContainAnyElementsOf(oldRecords);
+ oldTxn.abort();
+ newTxn.abort();
+ }
+
+ /**
+ * Publishes a message on {@code topic}, and acknowledges a message on {@code subscription} in {@code txn},
+ * waiting until the subscription has recorded it.
+ */
+ private void ackInTransaction(String topic, PulsarClient client, String subscription, Transaction txn)
+ throws Exception {
+ String segment = singleSegment(topic);
+ List<String> before = ackRecords(segment, subscription);
+ try (QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create()) {
+ producer.newMessage().key("k").value("acked").send();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message on %s", topic).isNotNull();
+ consumer.acknowledge(msg.id(), txn);
+ // The acknowledgment is sent asynchronously.
+ Awaitility.await().untilAsserted(() -> assertThat(ackRecords(segment, subscription))
+ .as("ack records of %s on %s", subscription, segment)
+ .hasSize(before.size() + 1));
+ }
+ }
+
+ /** The paths of the {@link TxnPaths#TXN_OP_PREFIX} ack records of {@code subscription} on the segment. */
+ private List<String> ackRecords(String segment, String subscription) throws Exception {
+ TxnMetadataStore txnStore = new TxnMetadataStore(getPulsar().getLocalMetadataStore());
+ List<String> paths = Collections.synchronizedList(new ArrayList<>());
+ txnStore.listAcksBySegmentSubscription(segment, subscription, 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 String singleSegment(String topic) throws Exception {
+ var segments = admin.scalableTopics().getMetadata(topic).getSegments().values().stream()
+ .filter(seg -> seg.isActive())
+ .toList();
+ assertThat(segments).as("active segments of %s", topic).hasSize(1);
+ var range = segments.get(0).getHashRange();
+ return SegmentTopicName.fromParent(TopicName.get(topic),
+ HashRange.of(range.getStart(), range.getEnd()), segments.get(0).getSegmentId()).toString();
+ }
+}
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
index a1734feec620..54642f1bcc95 100644
--- 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
@@ -50,8 +50,10 @@ import org.testng.annotations.Test;
* 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.
+ * open on the segment, which the buffer deletes as it applies their outcome, and for their ack records
+ * on the segment's subscriptions, which the pending-ack store of each subscription deletes the same way.
+ * Except while the segment is still active in the layout of its topic: a client can then load it again.
+ * Deleting just a subscription must delete its ack records too.
*/
public class ScalableTopicSegmentDeleteTxnStateTest extends V5ClientBaseTest {
@@ -220,6 +222,42 @@ public class ScalableTopicSegmentDeleteTxnStateTest extends V5ClientBaseTest {
txn.abort();
}
+ @Test
+ public void testDeleteScalableTopicDeletesAckRecordsOfOpenTransaction() throws Exception {
+ String topic = newScalableTopic(1);
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ PulsarClient client = newTxnClient();
+ Transaction txn = client.newTransaction();
+ ackInTransaction(topic, client, "sub", txn);
+
+ admin.scalableTopics().deleteScalableTopic(topic, true);
+
+ assertThat(ackRecords(segment, "sub"))
+ .as("ack records on %s of a transaction still open after deleting %s", segment, topic)
+ .isEmpty();
+ txn.abort();
+ }
+
+ @Test
+ public void testDeleteSubscriptionDeletesAckRecordsOfOpenTransaction() throws Exception {
+ String topic = newScalableTopic(1);
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ PulsarClient client = newTxnClient();
+ Transaction txn = client.newTransaction();
+ ackInTransaction(topic, client, "sub", txn);
+ ackInTransaction(topic, client, "sub2", txn);
+
+ admin.scalableTopics().deleteSubscription(topic, "sub");
+
+ assertThat(ackRecords(segment, "sub"))
+ .as("ack records on %s of a transaction still open after deleting the subscription", segment)
+ .isEmpty();
+ assertThat(ackRecords(segment, "sub2"))
+ .as("ack records on %s of the same transaction on another subscription", segment)
+ .hasSize(1);
+ txn.abort();
+ }
+
private PulsarClient newTxnClient() throws Exception {
return track(PulsarClient.builder()
.serviceUrl(getBrokerServiceUrl())
@@ -264,6 +302,27 @@ public class ScalableTopicSegmentDeleteTxnStateTest extends V5ClientBaseTest {
return txn;
}
+ /** Publishes a message on {@code topic}, and acknowledges a message on {@code subscription} in {@code txn}. */
+ private void ackInTransaction(String topic, PulsarClient client, String subscription, Transaction txn)
+ throws Exception {
+ String segment = segmentTopicName(topic, singleActiveSegmentId(topic));
+ try (QueueConsumer<String> consumer = client.newQueueConsumer(Schema.string())
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribe();
+ Producer<String> producer = client.newProducer(Schema.string()).topic(topic).create()) {
+ producer.newMessage().key("k").value("acked").send();
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message on %s", topic).isNotNull();
+ consumer.acknowledge(msg.id(), txn);
+ // The acknowledgment is sent asynchronously: wait until the subscription has recorded it.
+ Awaitility.await().untilAsserted(() -> assertThat(ackRecords(segment, subscription))
+ .as("ack records of %s on %s", subscription, segment)
+ .hasSize(1));
+ }
+ }
+
/** 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<>();
@@ -343,6 +402,27 @@ public class ScalableTopicSegmentDeleteTxnStateTest extends V5ClientBaseTest {
return paths;
}
+ /** The paths of the {@link TxnPaths#TXN_OP_PREFIX} ack records of {@code subscription} on the segment. */
+ private List<String> ackRecords(String segment, String subscription) throws Exception {
+ TxnMetadataStore txnStore = new TxnMetadataStore(getPulsar().getLocalMetadataStore());
+ List<String> paths = Collections.synchronizedList(new ArrayList<>());
+ txnStore.listAcksBySegmentSubscription(segment, subscription, 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);
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 38d0bbd36f7b..01de1ec7b9f4 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
@@ -25,15 +25,20 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Predicate;
import lombok.Cleanup;
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.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
+import org.apache.pulsar.metadata.api.Option;
import org.apache.pulsar.metadata.api.ScanConsumer;
import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore;
import org.awaitility.Awaitility;
import org.testng.annotations.Test;
@@ -50,6 +55,24 @@ public class TxnMetadataStoreTest {
MetadataStoreConfig.builder().fsyncEnable(false).build());
}
+ /**
+ * An in-memory store whose index scans return every record under the scanned prefix, like a store with a
+ * native index, which doesn't apply the scan's filter, matching more keys than asked for (Oxia's point lookups
+ * also match the keys that extend the one looked up).
+ */
+ private static MetadataStore newOverMatchingStore() throws Exception {
+ return new LocalMemoryMetadataStore("memory:local",
+ MetadataStoreConfig.builder().fsyncEnable(false).build()) {
+ @Override
+ protected CompletableFuture<Void> storeScanByIndex(String scanPathPrefix, String indexName,
+ String fromKeyInclusive, String toKeyInclusive,
+ Predicate<GetResult> fallbackFilter,
+ ScanConsumer consumer, Set<Option> opts) {
+ return storeScanChildren(scanPathPrefix, consumer, opts);
+ }
+ };
+ }
+
private static TxnHeader open(long createdMs, long timeoutMs) {
return new TxnHeader(TxnState.OPEN, Duration.ofMillis(timeoutMs),
Instant.ofEpochMilli(createdMs), null, null);
@@ -134,6 +157,101 @@ public class TxnMetadataStoreTest {
assertThat(remaining).containsExactlyInAnyOrder(otherSegmentWrite, ack);
}
+ @Test
+ public void deleteWriteOpsForSegment_keepsTheOtherRecordsAnIndexScanReturns() throws Exception {
+ @Cleanup MetadataStore store = newOverMatchingStore();
+ TxnMetadataStore txn = new TxnMetadataStore(store);
+
+ String seg = "segment://public/default/topic/0000-7fff-1";
+ // Its key extends the key of seg, so a point lookup on Oxia's index matches it too.
+ String extendingSeg = "segment://public/default/topic/0000-7fff-12";
+ txn.appendOp("tx-1", new TxnOp(TxnOpKind.WRITE, seg, null, 0L, 0L, null)).get();
+ TxnOp extendingSegmentWrite = new TxnOp(TxnOpKind.WRITE, extendingSeg, null, 0L, 0L, null);
+ txn.appendOp("tx-1", extendingSegmentWrite).get();
+ TxnOp ack = new TxnOp(TxnOpKind.ACK, seg, "sub", 1L, 5L, false);
+ txn.appendOp("tx-1", ack).get();
+
+ txn.deleteWriteOpsForSegment(seg).get();
+
+ List<TxnOp> remaining = new ArrayList<>();
+ txn.listOpsByTxn("tx-1", collectOps(remaining)).get();
+ assertThat(remaining).containsExactlyInAnyOrder(extendingSegmentWrite, ack);
+ }
+
+ @Test
+ public void collectAckOpsForSegmentSubscription_collectsTheSubscriptionsAckOpsOfEveryTxn() throws Exception {
+ @Cleanup MetadataStore store = newOverMatchingStore();
+ TxnMetadataStore txn = new TxnMetadataStore(store);
+
+ String seg = "segment://public/default/topic/0000-7fff-1";
+ String extendingSeg = "segment://public/default/topic/0000-7fff-12";
+ txn.appendOp("tx-1", new TxnOp(TxnOpKind.ACK, seg, "sub", 1L, 1L, false)).get();
+ txn.appendOp("tx-2", new TxnOp(TxnOpKind.ACK, seg, "sub", 1L, 2L, false)).get();
+ // Its key extends the key of (seg, "sub"), so a point lookup on Oxia's index matches it too.
+ TxnOp extendingSubscriptionAck = new TxnOp(TxnOpKind.ACK, seg, "sub2", 1L, 1L, false);
+ txn.appendOp("tx-1", extendingSubscriptionAck).get();
+ TxnOp extendingSegmentAck = new TxnOp(TxnOpKind.ACK, extendingSeg, "sub", 2L, 1L, false);
+ txn.appendOp("tx-1", extendingSegmentAck).get();
+ TxnOp write = new TxnOp(TxnOpKind.WRITE, seg, null, 0L, 0L, null);
+ txn.appendOp("tx-1", write).get();
+
+ List<TxnMetadataStore.OpRecord> ackOps = txn.collectAckOpsForSegmentSubscription(seg, "sub").get();
+ // Written after the collection, like the record of a subscription re-created with the same name.
+ TxnOp laterAck = new TxnOp(TxnOpKind.ACK, seg, "sub", 1L, 3L, false);
+ txn.appendOp("tx-3", laterAck).get();
+ txn.deleteOps(ackOps).get();
+
+ assertThat(ackOps).extracting(TxnMetadataStore.OpRecord::txnId).containsExactlyInAnyOrder("tx-1", "tx-2");
+ List<TxnOp> remaining = new ArrayList<>();
+ store.scanChildren(TxnPaths.TXN_OP_PREFIX, collectOps(remaining)).get();
+ assertThat(remaining)
+ .containsExactlyInAnyOrder(extendingSubscriptionAck, extendingSegmentAck, write, laterAck);
+ }
+
+ @Test
+ public void deleteOps_keepsARecordWrittenAtACollectedPathSince() throws Exception {
+ @Cleanup MetadataStore store = newMemoryStore();
+ TxnMetadataStore txn = new TxnMetadataStore(store);
+
+ String seg = "segment://public/default/topic/0000-7fff-1";
+ String path = txn.appendOp("tx-1", new TxnOp(TxnOpKind.ACK, seg, "sub", 1L, 1L, false)).get().getPath();
+ List<TxnMetadataStore.OpRecord> ackOps = txn.collectAckOpsForSegmentSubscription(seg, "sub").get();
+
+ // Another cleanup deletes the record, and the same txn acknowledges again on a subscription re-created
+ // with the same name: Oxia gives the new op the sequence key of the deleted one, with another version
+ // (versions only grow on Oxia; the in-memory store restarts a re-created path at 0, so update it once).
+ store.delete(path, Optional.empty()).get();
+ TxnOp newAck = new TxnOp(TxnOpKind.ACK, seg, "sub", 2L, 1L, false);
+ Stat created = store.put(path, TxnMetadataStore.toJson(newAck), Optional.of(-1L)).get();
+ store.put(path, TxnMetadataStore.toJson(newAck), Optional.of(created.getVersion())).get();
+ txn.deleteOps(ackOps).get();
+
+ List<TxnOp> remaining = new ArrayList<>();
+ store.scanChildren(TxnPaths.TXN_OP_PREFIX, collectOps(remaining)).get();
+ assertThat(remaining).containsExactly(newAck);
+ }
+
+ @Test
+ public void deleteAckOpsForSegment_removesTheAckOpsOfEverySubscription() throws Exception {
+ @Cleanup MetadataStore store = newOverMatchingStore();
+ TxnMetadataStore txn = new TxnMetadataStore(store);
+
+ String seg = "segment://public/default/topic/0000-7fff-1";
+ String extendingSeg = "segment://public/default/topic/0000-7fff-12";
+ txn.appendOp("tx-1", new TxnOp(TxnOpKind.ACK, seg, "sub-a", 1L, 1L, false)).get();
+ txn.appendOp("tx-2", new TxnOp(TxnOpKind.ACK, seg, "sub-b", 1L, 2L, true)).get();
+ TxnOp extendingSegmentAck = new TxnOp(TxnOpKind.ACK, extendingSeg, "sub-a", 2L, 1L, false);
+ txn.appendOp("tx-1", extendingSegmentAck).get();
+ TxnOp write = new TxnOp(TxnOpKind.WRITE, seg, null, 0L, 0L, null);
+ txn.appendOp("tx-1", write).get();
+
+ txn.deleteAckOpsForSegment(seg).get();
+
+ List<TxnOp> remaining = new ArrayList<>();
+ store.scanChildren(TxnPaths.TXN_OP_PREFIX, collectOps(remaining)).get();
+ assertThat(remaining).containsExactlyInAnyOrder(extendingSegmentAck, write);
+ }
+
@Test
public void appendAndScanAckOps() throws Exception {
@Cleanup MetadataStore store = newMemoryStore();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnPathsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnPathsTest.java
new file mode 100644
index 000000000000..ff84a82bcd2c
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/metadata/TxnPathsTest.java
@@ -0,0 +1,44 @@
+/*
+ * 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.transaction.metadata;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.List;
+import org.testng.annotations.Test;
+
+public class TxnPathsTest {
+
+ @Test
+ public void ackIndexSegmentBoundsCoverTheAckKeysOfTheSegmentOnly() {
+ String segment = "segment://public/default/topic/0000-7fff-1";
+ String lower = TxnPaths.ackIndexSegmentLowerBound(segment);
+ String upper = TxnPaths.ackIndexSegmentUpperBound(segment);
+ // Subscription names with the separator, with '~', and with characters that sort high or low.
+ for (String subscription : List.of("sub", "a", "zzz", "~", "~~", "sub:x", "sub/x", "a b", "100%")) {
+ assertThat(TxnPaths.ackIndexKey(segment, subscription))
+ .as("ack index key of subscription %s", subscription)
+ .isBetween(lower, upper);
+ }
+ // The segments of the topic with the same hash range whose names sort next to this one's, the first one
+ // extending it.
+ assertThat(TxnPaths.ackIndexKey("segment://public/default/topic/0000-7fff-12", "sub")).isLessThan(lower);
+ assertThat(TxnPaths.ackIndexKey("segment://public/default/topic/0000-7fff-0", "sub")).isLessThan(lower);
+ assertThat(TxnPaths.ackIndexKey("segment://public/default/topic/0000-7fff-2", "sub")).isGreaterThan(upper);
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStoreTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStoreTest.java
index e9994fd33142..e8388e2f8e05 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStoreTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/pendingack/impl/MetadataPendingAckStoreTest.java
@@ -19,9 +19,12 @@
package org.apache.pulsar.broker.transaction.pendingack.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.eq;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.time.Duration;
@@ -45,6 +48,7 @@ import org.apache.pulsar.broker.transaction.metadata.TxnState;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
import org.apache.pulsar.metadata.api.Stat;
import org.awaitility.Awaitility;
@@ -129,6 +133,50 @@ public class MetadataPendingAckStoreTest {
assertThat(hits.get(0).getCumulative()).isTrue();
}
+ @Test
+ public void closeAsync_waitsForTheAcksBeingAppended() throws Exception {
+ TxnMetadataStore slowStore = spy(txnStore);
+ CompletableFuture<Stat> appending = new CompletableFuture<>();
+ doReturn(appending).when(slowStore).appendOp(any(), any());
+ MetadataPendingAckStore ackStore = new MetadataPendingAckStore(subscription, slowStore);
+ ackStore.replayAsync(handle, null);
+ ackStore.recoveryFuture().get(5, TimeUnit.SECONDS);
+
+ CompletableFuture<Void> ack = ackStore.appendIndividualAck(new TxnID(1, 1), Collections.singletonList(
+ MutablePair.of(PositionFactory.create(3, 5), 1)));
+ CompletableFuture<Void> closed = ackStore.closeAsync();
+
+ // The ack records of a deleted subscription are deleted once it is closed: none must land after that.
+ assertThat(closed).as("close while an ack is being appended").isNotDone();
+ assertThatThrownBy(() -> ackStore.appendIndividualAck(new TxnID(1, 2), Collections.singletonList(
+ MutablePair.of(PositionFactory.create(3, 6), 1))).get())
+ .as("ack after the close")
+ .hasCauseInstanceOf(IllegalStateException.class);
+
+ appending.complete(null);
+ closed.get(5, TimeUnit.SECONDS);
+ ack.get(5, TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void closeAsync_completesWhenAnAckBeingAppendedFails() throws Exception {
+ TxnMetadataStore slowStore = spy(txnStore);
+ CompletableFuture<Stat> appending = new CompletableFuture<>();
+ doReturn(appending).when(slowStore).appendOp(any(), any());
+ MetadataPendingAckStore ackStore = new MetadataPendingAckStore(subscription, slowStore);
+ ackStore.replayAsync(handle, null);
+ ackStore.recoveryFuture().get(5, TimeUnit.SECONDS);
+
+ CompletableFuture<Void> ack = ackStore.appendIndividualAck(new TxnID(1, 1), Collections.singletonList(
+ MutablePair.of(PositionFactory.create(3, 5), 1)));
+ CompletableFuture<Void> closed = ackStore.closeAsync();
+ appending.completeExceptionally(new MetadataStoreException("append failed"));
+
+ closed.get(5, TimeUnit.SECONDS);
+ assertThatThrownBy(() -> ack.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(MetadataStoreException.class);
+ }
+
@Test
public void appendCommitMarkAndAbortMark_areNoOps() throws Exception {
MetadataPendingAckStore ackStore = new MetadataPendingAckStore(subscription, txnStore);