From 02f9707b81edb96cab84f3f362cee33d665e49fd Mon Sep 17 00:00:00 2001 From: penghui Date: Fri, 22 Oct 2021 21:14:38 +0800 Subject: [PATCH] #12429 only fixed the compactor skips data issue, but the normal reader/consumer also skips data while enabled read compacted data and read from the earliest position. --- .../pulsar/compaction/CompactedTopicImpl.java | 4 +--- .../pulsar/compaction/CompactedTopicTest.java | 15 +++++++++++++++ 2 files changed, 16 insertions(+), 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 index 4375188d0eed5..4bc166412d6d6 100644 --- 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 index adb6f469ca864..2dd6f8aa4dfeb 100644 --- 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 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(); } }