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/18 17:10:56 UTC
(pulsar) 01/04: [fix][broker] Fix readCompacted skipping tombstones within compaction horizon (#26190)
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 b9b3c15310643327e615f8d14877c059ee0aa0d2
Author: grishaf <gr...@drivenets.com>
AuthorDate: Sun Oct 18 06:24:27 2026 +0300
[fix][broker] Fix readCompacted skipping tombstones within compaction horizon (#26190)
Co-authored-by: Cursor <cu...@cursor.com>
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit a7a09c6947f357101d708cf5bc83de8b6d93a2d8)
---
.../AbstractDispatcherSingleActiveConsumer.java | 5 +
.../PersistentDispatcherSingleActiveConsumer.java | 12 ++-
.../pulsar/compaction/CompactedTopicUtils.java | 97 ++++++++++++++++--
.../pulsar/compaction/TopicCompactionService.java | 4 +
.../pulsar/compaction/CompactedTopicUtilsTest.java | 111 +++++++++++++++++++++
.../apache/pulsar/compaction/CompactionTest.java | 86 ++++++++++++++++
6 files changed, 303 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
index bc825607aaf9..ebfd4e15ee99 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java
@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import lombok.CustomLog;
import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
@@ -65,6 +66,9 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBas
private volatile int isClosed = FALSE;
protected boolean isFirstRead = true;
+ // For compacted reads: the position from which the subscription has been reading the original ledger, after its
+ // compacted snapshot. Later compactions must not move it back to the compacted ledger past that point.
+ protected Position originalLedgerReadPosition;
private static final int CONSUMER_CONSISTENT_HASH_REPLICAS = 100;
public AbstractDispatcherSingleActiveConsumer(SubType subscriptionType, int partitionIndex,
@@ -242,6 +246,7 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBas
if (consumers.isEmpty()) {
isFirstRead = true;
+ originalLedgerReadPosition = null;
}
consumers.add(consumer);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index aec19e8a96f8..4eb3b06799a6 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -32,6 +32,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import lombok.Getter;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -165,7 +166,7 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
}
private synchronized void readEntriesComplete(List<Entry> entries, Consumer readConsumer, long epoch,
- long readOpEpoch) {
+ long readOpEpoch, Position originalLedgerReadPosition) {
if (readOpEpoch != this.readOpEpoch) {
// Stale completion: this read was disowned by a redeliver-driven cursor rewind + re-arm
// (internalRedeliverUnacknowledgedMessages) that already issued a newer read. Clearing
@@ -187,6 +188,9 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
havePendingRead = false;
isFirstRead = false;
+ if (originalLedgerReadPosition != null) {
+ this.originalLedgerReadPosition = originalLedgerReadPosition;
+ }
if (readBatchSize < serviceConfig.getDispatcherMaxReadBatchSize()) {
int newReadBatchSize = Math.min(readBatchSize * 2, serviceConfig.getDispatcherMaxReadBatchSize());
@@ -392,12 +396,14 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
// https://github.com/apache/pulsar/issues/13690
final var epoch = consumer.readCompacted() ? DEFAULT_CONSUMER_EPOCH : consumer.getConsumerEpoch();
final CompletableFuture<List<Entry>> entriesFuture;
+ final AtomicReference<Position> originalLedgerRead = new AtomicReference<>();
if (consumer.readCompacted()) {
boolean readFromEarliest = isFirstRead && MessageId.earliest.equals(consumer.getStartMessageId())
&& (!cursor.isDurable() || cursor.getName().equals(Compactor.COMPACTION_SUBSCRIPTION)
|| hasValidMarkDeletePosition(cursor));
entriesFuture = CompactedTopicUtils.asyncReadCompactedEntries(topic.getTopicCompactionService(),
- cursor, messagesToRead, bytesToRead, topic.getMaxReadPosition(), readFromEarliest, true);
+ cursor, messagesToRead, bytesToRead, topic.getMaxReadPosition(), readFromEarliest, true,
+ originalLedgerReadPosition, originalLedgerRead::set);
} else {
// The continuation below dispatches to our executor. With inline read completion enabled,
// cache hits skip the managed-ledger executor handoff and its unrelated publishing work.
@@ -406,7 +412,7 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
}
entriesFuture.whenCompleteAsync((entries, e) -> {
if (e == null) {
- readEntriesComplete(entries, consumer, epoch, readOpEpoch);
+ readEntriesComplete(entries, consumer, epoch, readOpEpoch, originalLedgerRead.get());
} else {
readEntriesFailed(e, consumer, readOpEpoch);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
index fba12d5f29f8..f86d1f96a178 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
@@ -22,14 +22,17 @@ import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.bookkeeper.mledger.util.ManagedLedgerUtils.readEntries;
import static org.apache.bookkeeper.mledger.util.ManagedLedgerUtils.readEntriesWithSkipOrWait;
import com.google.common.annotations.Beta;
+import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.commons.collections4.CollectionUtils;
+import org.jspecify.annotations.Nullable;
public class CompactedTopicUtils {
@@ -37,6 +40,30 @@ public class CompactedTopicUtils {
public static CompletableFuture<List<Entry>> asyncReadCompactedEntries(
TopicCompactionService topicCompactionService, ManagedCursor cursor, int maxEntries,
long bytesToRead, Position maxReadPosition, boolean readFromEarliest, boolean wait) {
+ return asyncReadCompactedEntries(topicCompactionService, cursor, maxEntries, bytesToRead, maxReadPosition,
+ readFromEarliest, wait, null, null);
+ }
+
+ /**
+ * Read the next entries of a compacted-topic reader.
+ *
+ * <p>Positions up to the compaction horizon are read from the compacted ledger, later positions from the
+ * original ledger. A reader that has already moved on to the original ledger must stay there even when a later
+ * compaction moves the horizon past its read position, or it would skip the messages compaction removed since,
+ * tombstones included.
+ *
+ * @param originalLedgerReadPosition the position from which this reader has been reading the original ledger,
+ * or {@code null}. Positions at or after it are read from the original ledger
+ * while it is still retained there.
+ * @param onOriginalLedgerRead called with the position from which the reader moves on to the original ledger,
+ * when a read finishes the compacted snapshot or reads the original ledger; may be
+ * {@code null}
+ */
+ @Beta
+ public static CompletableFuture<List<Entry>> asyncReadCompactedEntries(
+ TopicCompactionService topicCompactionService, ManagedCursor cursor, int maxEntries,
+ long bytesToRead, Position maxReadPosition, boolean readFromEarliest, boolean wait,
+ @Nullable Position originalLedgerReadPosition, @Nullable Consumer<Position> onOriginalLedgerRead) {
Objects.requireNonNull(topicCompactionService);
Objects.requireNonNull(cursor);
checkArgument(maxEntries > 0);
@@ -48,17 +75,23 @@ public class CompactedTopicUtils {
readPosition = cursor.getReadPosition();
}
+ // Once retention has trimmed the original ledger at that position, only the compacted ledger has the data.
+ final Position originalLedgerLimit = originalLedgerReadPosition != null
+ && isRetained(cursor, originalLedgerReadPosition) ? originalLedgerReadPosition : null;
+ if (originalLedgerLimit != null && readPosition.compareTo(originalLedgerLimit) >= 0) {
+ return readOriginalLedger(cursor, maxEntries, bytesToRead, maxReadPosition, wait);
+ }
+
CompletableFuture<Position> lastCompactedPositionFuture = topicCompactionService.getLastCompactedPosition();
return lastCompactedPositionFuture.thenCompose(lastCompactedPosition -> {
if (lastCompactedPosition == null
|| readPosition.compareTo(
lastCompactedPosition.getLedgerId(), lastCompactedPosition.getEntryId()) > 0) {
- if (wait) {
- return readEntriesWithSkipOrWait(cursor, maxEntries, bytesToRead, maxReadPosition, null);
- } else {
- return readEntries(cursor, maxEntries, bytesToRead, maxReadPosition);
+ if (onOriginalLedgerRead != null) {
+ onOriginalLedgerRead.accept(cursor.getReadPosition());
}
+ return readOriginalLedger(cursor, maxEntries, bytesToRead, maxReadPosition, wait);
}
int numberOfEntriesToRead = cursor.applyMaxSizeCap(maxEntries, bytesToRead);
@@ -67,23 +100,69 @@ public class CompactedTopicUtils {
.thenApply(entries -> {
if (CollectionUtils.isEmpty(entries)) {
Position seekToPosition = lastCompactedPosition.getNext();
+ if (originalLedgerLimit != null && seekToPosition.compareTo(originalLedgerLimit) > 0) {
+ seekToPosition = originalLedgerLimit;
+ }
if (readPosition.compareTo(seekToPosition.getLedgerId(), seekToPosition.getEntryId()) > 0) {
seekToPosition = readPosition;
}
cursor.seek(seekToPosition);
+ notifyIfPastHorizon(seekToPosition, lastCompactedPosition, onOriginalLedgerRead);
return entries;
}
+ List<Entry> result = entries;
+ if (originalLedgerLimit != null
+ && entries.get(entries.size() - 1).getPosition().compareTo(originalLedgerLimit) >= 0) {
+ // The reader already went past this point on the original ledger: continue there.
+ result = new ArrayList<>(entries.size());
+ for (Entry entry : entries) {
+ if (entry.getPosition().compareTo(originalLedgerLimit) < 0) {
+ result.add(entry);
+ } else {
+ entry.release();
+ }
+ }
+ if (result.isEmpty()) {
+ cursor.seek(originalLedgerLimit, true);
+ return result;
+ }
+ }
+
long entriesSize = 0;
- for (Entry entry : entries) {
+ for (Entry entry : result) {
entriesSize += entry.getLength();
}
- cursor.updateReadStats(entries.size(), entriesSize);
+ cursor.updateReadStats(result.size(), entriesSize);
- Entry lastEntry = entries.get(entries.size() - 1);
- cursor.seek(lastEntry.getPosition().getNext(), true);
- return entries;
+ Position nextPosition = result.get(result.size() - 1).getPosition().getNext();
+ cursor.seek(nextPosition, true);
+ notifyIfPastHorizon(nextPosition, lastCompactedPosition, onOriginalLedgerRead);
+ return result;
});
});
}
+
+ private static CompletableFuture<List<Entry>> readOriginalLedger(ManagedCursor cursor, int maxEntries,
+ long bytesToRead, Position maxReadPosition,
+ boolean wait) {
+ if (wait) {
+ return readEntriesWithSkipOrWait(cursor, maxEntries, bytesToRead, maxReadPosition, null);
+ } else {
+ return readEntries(cursor, maxEntries, bytesToRead, maxReadPosition);
+ }
+ }
+
+ // The snapshot ends when the cursor moves past the horizon: the next read goes to the original ledger.
+ private static void notifyIfPastHorizon(Position nextReadPosition, Position lastCompactedPosition,
+ @Nullable Consumer<Position> onOriginalLedgerRead) {
+ if (onOriginalLedgerRead != null && nextReadPosition.compareTo(lastCompactedPosition) > 0) {
+ onOriginalLedgerRead.accept(nextReadPosition);
+ }
+ }
+
+ private static boolean isRetained(ManagedCursor cursor, Position position) {
+ Position firstPosition = cursor.getManagedLedger().getFirstPosition();
+ return firstPosition == null || position.compareTo(firstPosition) > 0;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
index 6153f4cfa896..1a97aa2b31e3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/TopicCompactionService.java
@@ -41,6 +41,10 @@ public interface TopicCompactionService extends AutoCloseable {
/**
* Read the compacted entries from the TopicCompactionService.
*
+ * <p>Compacted readers use it for positions up to {@link #getLastCompactedPosition()}. A reader that has already
+ * moved past that position keeps reading the original topic from there, even after a later compaction moves the
+ * last compacted position past it, so that it doesn't miss the messages that compaction removed since.
+ *
* @param startPosition the position to start reading from.
* @param numberOfEntriesToRead the maximum number of entries to read.
* @return a future that will be completed with the list of entries, this list can be null.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java
index 6973a5d0fd40..9570e4eb376d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.compaction;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -25,6 +26,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
@@ -76,4 +78,113 @@ public class CompactedTopicUtilsTest {
Assert.assertNull(throwableRef.get());
Assert.assertEquals(readPositionRef.get(), lastCompactedPosition.getNext());
}
+
+ @Test
+ public void testOriginalLedgerReaderIgnoresLaterCompaction() throws Exception {
+ // The reader went on to the original ledger at 1:91, then a compaction moved the horizon to 1:100.
+ TopicCompactionService service = mockService(PositionFactory.create(1, 100));
+ ManagedCursorImpl cursor = mockCursor(PositionFactory.create(1, 95), PositionFactory.create(1, -1));
+ Entry tombstone = mockEntry(PositionFactory.create(1, 95));
+ Mockito.doAnswer(invocation -> {
+ AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(2);
+ callback.readEntriesComplete(new ArrayList<>(List.of(tombstone)), invocation.getArgument(3));
+ return null;
+ }).when(cursor).asyncReadEntries(Mockito.anyInt(), Mockito.anyLong(), Mockito.any(), Mockito.any(),
+ Mockito.any());
+
+ // The transaction max read position still bounds the read.
+ Position maxReadPosition = PositionFactory.create(1, 97);
+ List<Entry> entries = CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 10, 1000,
+ maxReadPosition, false, false, PositionFactory.create(1, 91), null).get();
+
+ Assert.assertEquals(entries, List.of(tombstone));
+ Mockito.verify(cursor).asyncReadEntries(Mockito.eq(10), Mockito.eq(1000L), Mockito.any(), Mockito.any(),
+ Mockito.eq(maxReadPosition));
+ Mockito.verify(service, Mockito.never()).readCompactedEntries(Mockito.any(), Mockito.anyInt());
+ Mockito.verify(cursor, Mockito.never()).seek(Mockito.any());
+ Mockito.verify(cursor, Mockito.never()).seek(Mockito.any(), Mockito.anyBoolean());
+ }
+
+ @Test
+ public void testCompactedReadStopsAtOriginalLedgerReadPosition() throws Exception {
+ // The cursor was rewound below the position where the reader went on to the original ledger.
+ Position originalLedgerReadPosition = PositionFactory.create(1, 91);
+ TopicCompactionService service = mockService(PositionFactory.create(1, 100));
+ ManagedCursorImpl cursor = mockCursor(PositionFactory.create(1, 80), PositionFactory.create(1, -1));
+ Entry before = mockEntry(PositionFactory.create(1, 85));
+ Entry after = mockEntry(PositionFactory.create(1, 95));
+ Mockito.doReturn(CompletableFuture.completedFuture(new ArrayList<>(List.of(before, after))))
+ .when(service).readCompactedEntries(Mockito.any(), Mockito.anyInt());
+
+ List<Entry> entries = CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 10, 1000,
+ PositionFactory.LATEST, false, false, originalLedgerReadPosition, null).get();
+
+ Assert.assertEquals(entries, List.of(before));
+ Mockito.verify(after).release();
+ Mockito.verify(before, Mockito.never()).release();
+ Mockito.verify(cursor).seek(PositionFactory.create(1, 86), true);
+ }
+
+ @Test
+ public void testTrimmedOriginalLedgerFallsBackToCompactedLedger() throws Exception {
+ // Retention deleted ledger 1, so the compacted ledger is the only copy of the data at 1:95.
+ TopicCompactionService service = mockService(PositionFactory.create(3, 0));
+ ManagedCursorImpl cursor = mockCursor(PositionFactory.create(1, 95), PositionFactory.create(2, -1));
+ Entry compacted = mockEntry(PositionFactory.create(2, 5));
+ Mockito.doReturn(CompletableFuture.completedFuture(new ArrayList<>(List.of(compacted))))
+ .when(service).readCompactedEntries(Mockito.any(), Mockito.anyInt());
+
+ List<Entry> entries = CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 10, 1000,
+ PositionFactory.LATEST, false, false, PositionFactory.create(1, 91), null).get();
+
+ Assert.assertEquals(entries, List.of(compacted));
+ Mockito.verify(cursor).seek(PositionFactory.create(2, 6), true);
+ }
+
+ @Test
+ public void testEndOfCompactedSnapshotIsReported() throws Exception {
+ Position lastCompactedPosition = PositionFactory.create(1, 100);
+ TopicCompactionService service = mockService(lastCompactedPosition);
+ ManagedCursorImpl cursor = mockCursor(PositionFactory.create(1, 80), PositionFactory.create(1, -1));
+ Entry last = mockEntry(lastCompactedPosition);
+ Mockito.doReturn(CompletableFuture.completedFuture(new ArrayList<>(List.of(last))))
+ .when(service).readCompactedEntries(Mockito.any(), Mockito.anyInt());
+
+ AtomicReference<Position> reported = new AtomicReference<>();
+ CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 10, 1000, PositionFactory.LATEST, false,
+ false, null, reported::set).get();
+ Assert.assertEquals(reported.get(), lastCompactedPosition.getNext());
+
+ // No surviving entry left before the horizon: the cursor skips to the original ledger.
+ Mockito.doReturn(CompletableFuture.completedFuture(Collections.emptyList()))
+ .when(service).readCompactedEntries(Mockito.any(), Mockito.anyInt());
+ reported.set(null);
+ CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 10, 1000, PositionFactory.LATEST, false,
+ false, null, reported::set).get();
+ Assert.assertEquals(reported.get(), lastCompactedPosition.getNext());
+ }
+
+ private static TopicCompactionService mockService(Position lastCompactedPosition) {
+ TopicCompactionService service = Mockito.mock(TopicCompactionService.class);
+ Mockito.doReturn(CompletableFuture.completedFuture(lastCompactedPosition)).when(service)
+ .getLastCompactedPosition();
+ return service;
+ }
+
+ private static ManagedCursorImpl mockCursor(Position readPosition, Position firstLedgerPosition) {
+ ManagedLedger managedLedger = Mockito.mock(ManagedLedger.class);
+ Mockito.doReturn(firstLedgerPosition).when(managedLedger).getFirstPosition();
+ ManagedCursorImpl cursor = Mockito.mock(ManagedCursorImpl.class);
+ Mockito.doReturn(managedLedger).when(cursor).getManagedLedger();
+ Mockito.doReturn(readPosition).when(cursor).getReadPosition();
+ Mockito.doReturn(10).when(cursor).applyMaxSizeCap(Mockito.anyInt(), Mockito.anyLong());
+ return cursor;
+ }
+
+ private static Entry mockEntry(Position position) {
+ Entry entry = Mockito.mock(Entry.class);
+ Mockito.doReturn(position).when(entry).getPosition();
+ Mockito.doReturn(10).when(entry).getLength();
+ return entry;
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index 02165fd9ba14..272437d82113 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -2791,6 +2791,92 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
admin.topics().compactionStatus(topic).status, LongRunningProcessStatus.Status.SUCCESS));
}
+ /**
+ * A readCompacted reader that already finished the compacted snapshot must receive the tombstones published
+ * afterwards, even when a second compaction removes them before they are read. A reader created after that
+ * compaction still sees only the compacted view.
+ */
+ @Test(timeOut = 30000)
+ public void testReaderReceivesTombstonesWhenCompactionRunsBeforeConsumption() throws Exception {
+ admin.namespaces().setRetention("my-tenant/my-ns", new RetentionPolicies(-1, -1));
+ final String topic = newUniqueName("persistent://my-tenant/my-ns/tombstones-after-snapshot");
+
+ pulsarClient.newConsumer().topic(topic).subscriptionName("sub1")
+ .readCompacted(true).subscribe().close();
+
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .enableBatching(false)
+ .create();
+
+ final String keptKey = "key-keep";
+ final List<String> deletedKeys = List.of("key-1", "key-2", "key-3");
+
+ producer.newMessage().key(keptKey).value("value-keep").send();
+ for (String key : deletedKeys) {
+ producer.newMessage().key(key).value("value-" + key).send();
+ }
+ triggerAndWaitCompaction(topic);
+
+ // A receiver queue of 1 keeps all but the first tombstone in the topic until the reader asks for them.
+ @Cleanup
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .readCompacted(true)
+ .receiverQueueSize(1)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ Map<String, String> receivedValues = new HashMap<>();
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ receivedValues.put(message.getKey(), message.getValue());
+ }
+ assertEquals(receivedValues, Map.of(
+ keptKey, "value-keep",
+ "key-1", "value-key-1",
+ "key-2", "value-key-2",
+ "key-3", "value-key-3"));
+
+ MessageIdAdv lastTombstoneId = null;
+ for (String key : deletedKeys) {
+ lastTombstoneId = (MessageIdAdv) producer.newMessage().key(key).send();
+ }
+ final Position lastTombstone = PositionFactory.create(lastTombstoneId.getLedgerId(),
+ lastTombstoneId.getEntryId());
+ final PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).orElseThrow();
+ admin.topics().triggerCompaction(topic);
+ Awaitility.await().untilAsserted(() -> {
+ Position horizon = persistentTopic.getTopicCompactionService().getLastCompactedPosition().get();
+ assertNotNull(horizon);
+ assertTrue(horizon.compareTo(lastTombstone) >= 0);
+ });
+
+ final List<String> receivedTombstones = new ArrayList<>();
+ for (int i = 0; i < deletedKeys.size(); i++) {
+ Message<String> message = reader.readNext(5, TimeUnit.SECONDS);
+ assertNotNull(message, "Expected tombstone " + (i + 1) + " of " + deletedKeys.size());
+ assertNull(message.getValue(), "Expected a tombstone for key " + message.getKey());
+ receivedTombstones.add(message.getKey());
+ }
+ assertEquals(receivedTombstones, deletedKeys);
+ assertNull(reader.readNext(1, TimeUnit.SECONDS));
+
+ @Cleanup
+ Reader<String> freshReader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .readCompacted(true)
+ .startMessageId(MessageId.earliest)
+ .create();
+ Message<String> message = freshReader.readNext(5, TimeUnit.SECONDS);
+ assertNotNull(message);
+ assertEquals(message.getKey(), keptKey);
+ assertEquals(message.getValue(), "value-keep");
+ assertNull(freshReader.readNext(1, TimeUnit.SECONDS));
+ }
+
@Test
public void testReaderReadOnDeletedLedger() throws Exception {
final var topic = "persistent://my-tenant/my-ns/reader-read-on-deleted-ledger";