Re: [PR] [fix][broker] Fix readCompacted skipping tombstones within compaction horizon [pulsar]

["Denovo1998 (via GitHub)" <gi...@apache.org>]

Denovo1998 commented on code in PR #26190:
URL: https://github.com/apache/pulsar/pull/26190#discussion_r3622775378


##########
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;
                         }

Review Comment:
   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.



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

Review Comment:
   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.



##########
pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java:
##########
@@ -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 {

Review Comment:
   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.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org