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

pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicUtils.java

@@ -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;
                         }
 
                         long entriesSize = 0;
@@ -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);
                     });
         });
     }

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java

@@ -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());
         }
         Assert.assertFalse(reader.hasMessageAvailable());
         Assert.assertNull(reader.readNext(3, TimeUnit.SECONDS));

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicUtilsTest.java

@@ -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;
@@ -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());
@@ -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());
     }
 }

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java

@@ -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;
@@ -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 {
+        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";