Skip to content

[fix][broker] Fix readCompacted skipping tombstones within compaction horizon #26190

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
grishaf wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[fix][broker] Fix readCompacted skipping tombstones within compaction horizon #26190

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ddb3ee1
Fix readCompacted skipping tombstones within compaction horizon
grishaf Jul 14, 2026
7e038f8
Merge upstream/master into fix/readcompacted-tombstone-skip
grishaf Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,14 +64,14 @@ public static CompletableFuture<List<Entry>> asyncReadCompactedEntries(
int numberOfEntriesToRead = cursor.applyMaxSizeCap(maxEntries, bytesToRead);

return topicCompactionService.readCompactedEntries(readPosition, numberOfEntriesToRead)
.thenApply(entries -> {
.thenCompose(entries -> {
if (CollectionUtils.isEmpty(entries)) {
Position seekToPosition = lastCompactedPosition.getNext();
if (readPosition.compareTo(seekToPosition.getLedgerId(), seekToPosition.getEntryId()) > 0) {
seekToPosition = readPosition;
if (wait) {
return readEntriesWithSkipOrWait(cursor, maxEntries, bytesToRead,
maxReadPosition, null);
} else {
return readEntries(cursor, maxEntries, bytesToRead, maxReadPosition);
}
cursor.seek(seekToPosition);
return entries;
}
Comment on lines 68 to 75

@Denovo1998 Denovo1998 Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

An empty compacted read is also the normal end-of-snapshot signal, and readCompactedEntries returns an empty list for COMPACT_LEDGER_EMPTY, NEWER_THAN_COMPACTED, and NoSuchElementException. It does not necessarily mean that this consumer has already transitioned to the original-ledger tail.

Falling back to readEntries here makes a newly created readCompacted reader replay entries that compaction intentionally removed. The modified test demonstrates this: after reading keys 0–4 from the compacted ledger, the reader now receives the pre-delete values and tombstones for keys 5–9. This contradicts the ReaderBuilder.readCompacted contract, which says that only the latest value for each key should be exposed up to the compaction horizon.

Could we preserve the initial compacted-snapshot behavior and track an explicit snapshot/tail phase for the dispatcher? Once a reader has completed its initial snapshot, it should remain on the original ledger even when a later compaction advances the global horizon.


long entriesSize = 0;
Expand All @@ -82,7 +82,7 @@ public static CompletableFuture<List<Entry>> asyncReadCompactedEntries(

Entry lastEntry = entries.get(entries.size() - 1);
cursor.seek(lastEntry.getPosition().getNext(), true);
return entries;
return CompletableFuture.completedFuture(entries);
});
});
}
Expand Down
14 changes: 12 additions & 2 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -730,8 +730,18 @@ public void testHasMessageAvailableWithNullValueMessage() throws Exception {
.startMessageId(MessageId.earliest)
.readCompacted(true)
.create();
for (int i = numMessages / 2; i < numMessages; ++i) {
reader.readNext();
for (int i = 0; i < numMessages / 2; ++i) {
Message<byte[]> message = reader.readNext();
Assert.assertEquals(message.getKey(), String.valueOf(i));
Assert.assertEquals(new String(message.getData()), String.format("msg [%d]", i));
}
// After the compacted snapshot, remaining entries within the compaction horizon are
// read from the original ledger (pre-delete values and tombstones for compacted-out keys).
Message<byte[]> message;
while ((message = reader.readNext(1, TimeUnit.SECONDS)) != null) {
int key = Integer.parseInt(message.getKey());
Assert.assertTrue(key >= numMessages / 2 && key < numMessages,
"Unexpected key after compacted snapshot: " + message.getKey());
Comment on lines +733 to +744

@Denovo1998 Denovo1998 Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This changes the expected readCompacted semantics rather than only adding coverage for the continuous-reader race.

A fresh reader created after compaction should still see only keys 0–4. Accepting the pre-delete values and tombstones from the original ledger masks the regression introduced by the production change.

Please keep the original fresh-reader expectation and add a separate test where the reader has already completed the initial compacted snapshot before the tombstones are published and the second compaction runs.

Also, this loop only checks the key range. It does not verify the exact number of messages, ordering, duplicate delivery, or whether the messages are actually tombstones.

}
Assert.assertFalse(reader.hasMessageAvailable());
Assert.assertNull(reader.readNext(3, TimeUnit.SECONDS));
Expand Down
32 changes: 11 additions & 21 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
Expand All @@ -38,15 +37,14 @@ public class CompactedTopicUtilsTest {
@Test
public void testReadCompactedEntriesWithEmptyEntries() throws ExecutionException, InterruptedException {
Position lastCompactedPosition = PositionFactory.create(1, 100);
Position readPosition = PositionFactory.create(1, 91);
TopicCompactionService service = Mockito.mock(TopicCompactionService.class);
Mockito.doReturn(CompletableFuture.completedFuture(Collections.emptyList()))
.when(service).readCompactedEntries(Mockito.any(), Mockito.intThat(argument -> argument > 0));
Mockito.doReturn(CompletableFuture.completedFuture(lastCompactedPosition)).when(service)
.getLastCompactedPosition();


Position initPosition = PositionFactory.create(1, 90);
AtomicReference<Position> readPositionRef = new AtomicReference<>(initPosition.getNext());
AtomicReference<Position> readPositionRef = new AtomicReference<>(readPosition);
ManagedCursorImpl cursor = Mockito.mock(ManagedCursorImpl.class);
Mockito.doReturn(readPositionRef.get()).when(cursor).getReadPosition();
Mockito.doReturn(1).when(cursor).applyMaxSizeCap(Mockito.anyInt(), Mockito.anyLong());
Expand All @@ -55,25 +53,17 @@ public void testReadCompactedEntriesWithEmptyEntries() throws ExecutionException
return null;
}).when(cursor).seek(Mockito.any());

CompletableFuture<List<Entry>> completableFuture = new CompletableFuture<>();
final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
AsyncCallbacks.ReadEntriesCallback readEntriesCallback = new AsyncCallbacks.ReadEntriesCallback() {
@Override
public void readEntriesComplete(List<Entry> entries, Object ctx) {
completableFuture.complete(entries);
}

@Override
public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
completableFuture.completeExceptionally(exception);
throwableRef.set(exception);
}
};
Mockito.doAnswer(invocation -> {
AsyncCallbacks.ReadEntriesCallback callback = invocation.getArgument(2);
callback.readEntriesComplete(Collections.emptyList(), null);
return null;
}).when(cursor).asyncReadEntries(Mockito.anyInt(), Mockito.anyLong(), Mockito.any(),
Mockito.any(), Mockito.any());

final var entries = CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 1, 100,
final List<Entry> entries = CompactedTopicUtils.asyncReadCompactedEntries(service, cursor, 1, 100,
PositionFactory.LATEST, false, false).get();
Assert.assertTrue(entries.isEmpty());
Assert.assertNull(throwableRef.get());
Assert.assertEquals(readPositionRef.get(), lastCompactedPosition.getNext());
Assert.assertEquals(readPositionRef.get(), readPosition);
Mockito.verify(cursor, Mockito.never()).seek(lastCompactedPosition.getNext());
}
}
67 changes: 67 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -2781,6 +2782,72 @@ private void triggerAndWaitCompaction(String topic) throws Exception {
admin.topics().compactionStatus(topic).status, LongRunningProcessStatus.Status.SUCCESS));
}

/**
* Verifies that a continuous readCompacted reader receives tombstones published after the
* initial compaction snapshot, even when a second compaction runs before those tombstones
* are consumed.
*/
@Test(timeOut = 15000)
public void testTableViewMissesDeletesWhenCompactionRunsBeforeTombstonesConsumed() throws Exception {
Comment on lines +2785 to +2791

@Denovo1998 Denovo1998 Jul 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The test name and Javadoc describe a TableView regression, but the test only creates a Reader.

Could this either be renamed to describe the continuous readCompacted reader behavior, or use an actual TableView and verify that the deleted keys are removed from the map and that listeners receive the null-value updates? That would directly cover the user-visible impact described in the PR.

admin.namespaces().setRetention("my-tenant/my-ns", new RetentionPolicies(-1, -1));
final String topic = "persistent://my-tenant/my-ns/tableview-delete-after-compaction";

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);

@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"));

for (String key : deletedKeys) {
producer.newMessage().key(key).send();
}
triggerAndWaitCompaction(topic);

final Set<String> receivedTombstones = new HashSet<>();
for (int i = 0; i < deletedKeys.size(); i++) {
Message<String> message = reader.readNext(1, TimeUnit.SECONDS);
assertNotNull(message, "Expected tombstone " + (i + 1) + " of " + deletedKeys.size());
assertTrue(message.hasKey());
assertNull(message.getValue(), "Expected tombstone for key: " + message.getKey());
assertTrue(deletedKeys.contains(message.getKey()),
"Unexpected tombstone key: " + message.getKey());
receivedTombstones.add(message.getKey());
}
assertEquals(receivedTombstones, Set.copyOf(deletedKeys),
"Should receive a tombstone for each deleted key");
}

@Test
public void testReaderReadOnDeletedLedger() throws Exception {
final var topic = "persistent://my-tenant/my-ns/reader-read-on-deleted-ledger";
Expand Down