apache#12429 only fixed the compactor skips data issue, but the norma…

…l reader/consumer (apache#12464) also skips data while enabled read compacted data and read from the earliest position.
eolivelli · Oct 24, 2021 · dd90657 · dd90657
1 parent 86f40c1
commit dd90657
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 3 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -18,7 +18,6 @@
  */
 package org.apache.pulsar.compaction;
 
-import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
 import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.google.common.collect.ComparisonChain;
@@ -128,8 +127,7 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor,
                                         // need to force seek the read position to ensure the compactor can read
                                         // the complete last snapshot because of the compactor will read the data
                                         // before the compaction cursor mark delete position
-                                        cursor.seek(lastEntry.getPosition().getNext(),
-                                                cursor.getName().equals(COMPACTION_SUBSCRIPTION));
+                                        cursor.seek(lastEntry.getPosition().getNext(), true);
                                         callback.readEntriesComplete(entries, consumer);
                                     });
                             }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -481,5 +481,20 @@ public void testDoNotLossTheLastCompactedLedgerData() throws Exception {
             PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
             Assert.assertEquals(stats.compactedLedger.entries, keys + 1);
         });
+
+        // Make sure the reader can get all data from the compacted ledger and original ledger.
+        Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+                .topic(topic)
+                .startMessageId(MessageId.earliest)
+                .readCompacted(true)
+                .create();
+        int received = 0;
+        while (reader.hasMessageAvailable()) {
+            reader.readNext();
+            received++;
+        }
+        Assert.assertEquals(received, keys + 1);
+        reader.close();
+        producer.close();
     }
 }