diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index 58fcff877ca6c..0e465042ae07c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -2602,7 +2602,7 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture promise) { ls.getLedgerId()); } break; - } + }m // if truncate, all ledgers besides currentLedger are going to be deleted if (isTruncate) { if (log.isDebugEnabled()) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java index a13dd0499a6bc..d4223d3577f82 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java @@ -33,6 +33,7 @@ import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.transaction.TxnID; +import org.apache.pulsar.client.impl.ReaderImpl; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.util.FutureUtil; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReproduceCantGetMessageButHasMessageAvailableTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReproduceCantGetMessageButHasMessageAvailableTest.java new file mode 100644 index 0000000000000..36c482ddbb585 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReproduceCantGetMessageButHasMessageAvailableTest.java @@ -0,0 +1,439 @@ +package org.apache.pulsar.client.api; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.impl.ConsumerImpl; +import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.compaction.Compactor; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/*** + * TODO discuss: + * plan-1: 发现读到 null 就 reset last message id in broker。这将导致 read timeout 被当作一种正常现象。 + * plan-2: 增加 api: hasMessageAvailable(boolean fetchLastMessageIdFromBroker) + * plan-3: 增加 cmd，发生了 lastMessageId 变小的了情况，通知给 client。 + * plan-4: last message id 记录在 broker.consumer. 用于改变 compaction 的行为。[重要的] + * + * TODO Discuss 2: + * retention 策略不是会把 cursor 往前移动吗？ + * compaction 最后一条消息不执行 compaction + * + * TODO Enabled batch 后的 latLastMessageId 更新。 + * + * TODO + * 1.增加可选项：避免 trim 正在读取的 ledger 【重要的】 + * 2.compaction 的问题单独解决 + * 3.transaction buffer 单独解决 + * + * 1: 决定留哪些 entry + * 2：compaction cursor markDeleted 往前挪一下（ 如果有 reader 还有 race condition ）。 + * + * task running 参数：是否执行最后一条。 + * reader + * + * This test is just to show that reader may not consume any messages even if "reader.hasMessageAvailable" returns true. + */ +public class ReproduceCantGetMessageButHasMessageAvailableTest extends ProducerConsumerBase { + + private static final String NAMESPACE = "public/default"; + + @BeforeClass + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + + } + + @AfterClass(alwaysRun = true) + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + // Disable the scheduled task: compaction. + conf.setBrokerServiceCompactionMonitorIntervalInSeconds(Integer.MAX_VALUE); + // Disable the scheduled task: retention. + conf.setRetentionCheckIntervalInSeconds(Integer.MAX_VALUE); + // Messages can be held for a maximum of one minute. + conf.setDefaultRetentionTimeInMinutes(120); + } + + private String randomTopic(){ + return String.format("persistent://%s/%s", NAMESPACE, BrokerTestUtil.newUniqueName("tp")); + } + + /** + * Enabled compacted read and without batch sends, the last message has been deleted by compaction task. + */ + @Test + public void testRaceConditionWithCompactionDisabledBatch() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + List> messagesToSend = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k2", null), + Pair.of("k3", "v3"), + Pair.of("k3", null) + ); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .create(); + + sendMessages(topicName, false, messagesToSend, 1); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + reader.hasMessageAvailable(); + + // The last message id of compacted topic will less than `managedLedger.lastConfirmedPosition`. + triggerCompactionAndWait(topicName); + + verifyHasMessageAvailableButCantGet(reader); + + // cleanup. + reader.close(); + admin.topics().delete(topicName, false); + } + + /** + * Enabled compacted read and with batch sends, the last message has been deleted by compaction task. + */ + @Test + public void testRaceConditionWithCompactionEnabledBatch() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + List> firstEntry = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2") + ); + List> secondEntry = Arrays.asList( + Pair.of("k3", "v3"), + Pair.of("k4", "v4"), + Pair.of("k4", null) + ); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .create(); + + sendMessages(topicName, true, firstEntry, 1); + sendMessages(topicName, true, secondEntry, 1); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + reader.hasMessageAvailable(); + + // The last message id of compacted topic will equals `managedLedger.lastConfirmedPosition`. But the last + // message in the last entry will be lost by consumer, because it has been marked `compactedOut`. + triggerCompactionAndWait(topicName); + + verifyHasMessageAvailableButCantGet(reader); + + // cleanup. + reader.close(); + admin.topics().delete(topicName, false); + } + + /** + * Enabled compacted read and with batch sends, the last message read from compacted topic and marked "compactedOut" + * has been lost by consumer. + */ + @Test + public void testReadAfterCompactionEnabledBatch() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + List> messagesToSend = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k2", null), + Pair.of("k3", "v3"), + Pair.of("k3", null) + ); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .create(); + + sendMessages(topicName, true, messagesToSend, 3); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + triggerCompactionAndWait(topicName); + + // The variable `lastMessageIdInBroker` of consumer will be the last message id of compacted topic. But the last + // message in the last entry will be lost by consumer, because it has been marked `compactedOut`. + verifyHasMessageAvailableButCantGet(reader); + + // cleanup. + reader.close(); + admin.topics().delete(topicName, false); + } + + /** + * No durable cursor exists, all messages deleted by trim ledgers task. + */ + @Test + public void testRaceConditionWithRetentionAndNoDurableCursorExists() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + List> messagesToSend = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k3", "v3"), + Pair.of("k4", "v4") + ); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(false) + .create(); + + sendMessages(topicName, false, messagesToSend, 1); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + reader.hasMessageAvailable(); + + triggerLedgerSwitch(topicName); + clearAllTheLedgersOutdated(topicName); + + verifyHasMessageAvailableButCantGet(reader); + + // cleanup. + reader.close(); + admin.topics().delete(topicName, false); + } + + /** + * There are durable cursors, and durable cursors have been consumed to the end. The last non-empty ledger has been + * deleted when opening managed ledger. + */ + @Test + public void testRaceConditionWithForwardCursorWhenOpenManagedLedger() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + String subNameDurable = "sub_durable"; + List> messagesToSend = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k3", "v3"), + Pair.of("k4", "v4") + ); + ConsumerImpl consumer = (ConsumerImpl) pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .consumerName("c_reproduce_issue") + .subscriptionName(subNameDurable) + .receiverQueueSize(1000) + .readCompacted(false) + .subscribe(); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(false) + .create(); + + sendMessages(topicName, false, messagesToSend, 1); + + ackAllMessages(consumer); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + reader.hasMessageAvailable(); + + triggerLedgerSwitch(topicName); + clearAllTheLedgersOutdated(topicName); + + verifyHasMessageAvailableButCantGet(reader); + + // cleanup. + consumer.close(); + reader.close(); + admin.topics().delete(topicName, false); + } + + /** + * There are durable cursors, and durable cursors have been consumed to the end. If the managed ledger is not closed + * and then reopened, the last non-empty ledger will not be deleted. + * This test is used only to demonstrate the existence of a phenomenon, and has no other purpose. + */ + @Test(timeOut = 180 * 1000) + public void testRaceConditionWithRetentionAndOneDurableCursorExists() throws Exception { + String topicName = randomTopic(); + String subName = "sub_no_durable"; + String subNameDurable = "sub_durable"; + List> messagesToSend = Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k3", "v3"), + Pair.of("k4", "v4") + ); + ConsumerImpl consumer = (ConsumerImpl) pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .consumerName("c_reproduce_issue") + .subscriptionName(subNameDurable) + .receiverQueueSize(1000) + .readCompacted(false) + .subscribe(); + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .readerName("reproduce_issue") + .startMessageId(MessageId.earliest) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(false) + .create(); + + sendMessages(topicName, false, messagesToSend, 1); + + // The variable `lastMessageIdInBroker` of consumer will be `managedLedger.lastConfirmedPosition`. + reader.hasMessageAvailable(); + + triggerLedgerSwitch(topicName); + + ackAllMessages(consumer); + + setRetentionTimeMillis(topicName, 1); + try { + clearAllTheLedgersOutdated(topicName); + fail("Confirm that the retention policy does not delete the last non-empty ledger, even if it has been" + + " consumed"); + } catch (Exception ex){ + // ignore. + } + + // cleanup. + consumer.close(); + reader.close(); + admin.topics().delete(topicName, false); + } + + private void setRetentionTimeMillis(String topicName, int RetentionTimeMillis) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + managedLedger.getConfig().setRetentionTime(RetentionTimeMillis, TimeUnit.MILLISECONDS); + } + + private void ackAllMessages(ConsumerImpl consumer) throws Exception { + Message message = null; + while ((message = consumer.receive(2, TimeUnit.SECONDS)) != null){ + consumer.acknowledgeCumulative(message); + } + } + + private void triggerLedgerSwitch(String topicName) throws Exception{ + admin.topics().unload(topicName); + Awaitility.await().until(() -> { + CompletableFuture> topicFuture = + pulsar.getBrokerService().getTopic(topicName, false); + if (!topicFuture.isDone() || topicFuture.isCompletedExceptionally()){ + return false; + } + Optional topicOptional = topicFuture.join(); + if (!topicOptional.isPresent()){ + return false; + } + PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get(); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + return managedLedger.getState() == ManagedLedgerImpl.State.LedgerOpened; + }); + } + + private void clearAllTheLedgersOutdated(String topicName) throws Exception{ + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> { + CompletableFuture future = new CompletableFuture(); + managedLedger.trimConsumedLedgersInBackground(future); + future.join(); + return managedLedger.getLedgersInfo().size() == 1; + }); + } + + private void verifyHasMessageAvailableButCantGet(Reader reader) throws Exception { + boolean receiveNullEvenIfHasMessageAvailable = false; + while (reader.hasMessageAvailable()) { + Message message = reader.readNext(2, TimeUnit.SECONDS); + if (message == null) { + receiveNullEvenIfHasMessageAvailable = true; + break; + } + } + assertTrue(receiveNullEvenIfHasMessageAvailable, "If this test fails, you need to modify the doc for" + + " method hasMessageAvailable。"); + } + + private void triggerCompactionAndWait(String topicName) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + persistentTopic.triggerCompaction(); + + Awaitility.await().untilAsserted(() -> { + PositionImpl lastConfirmPos = (PositionImpl) persistentTopic.getManagedLedger().getLastConfirmedEntry(); + PositionImpl markDeletePos = (PositionImpl) persistentTopic + .getSubscription(Compactor.COMPACTION_SUBSCRIPTION).getCursor().getMarkDeletedPosition(); + assertEquals(markDeletePos.getLedgerId(), lastConfirmPos.getLedgerId()); + assertEquals(markDeletePos.getEntryId(), lastConfirmPos.getEntryId()); + }); + } + + private void sendMessages(String topicName, boolean enabledBatch, List> messagesToSend, + int sendMessagesLoopCount) throws Exception { + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(enabledBatch) + .create(); + + List> sendFutures = new ArrayList<>(); + for (int i = 0; i < sendMessagesLoopCount; i++) { + for (Pair messageToSend : messagesToSend) { + String key = messageToSend.getLeft(); + String value = messageToSend.getRight(); + if (key == null) { + sendFutures.add(producer.newMessage().value(value).sendAsync()); + } else { + sendFutures.add(producer.newMessage().key(key).value(value).sendAsync()); + } + } + producer.flush(); + } + FutureUtil.waitForAll(sendFutures).join(); + producer.close(); + } +} \ No newline at end of file diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java index 681b4a39c8e25..3aa75e0863196 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java @@ -34,6 +34,8 @@ import java.nio.file.Files; import java.nio.file.Paths; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -54,6 +56,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.ManagedLedgerInfo; import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; @@ -72,6 +75,7 @@ import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Reader; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.BatchMessageIdImpl; @@ -321,6 +325,227 @@ public void testReadEntriesAfterCompaction() throws Exception { } } + @DataProvider(name = "messagesToSend") + public Object[][] messagesToSend() { + return new Object[][]{ + // no message. + {Collections.emptyList()}, + // message without key. + { Arrays.asList( + Pair.of(null, "v"), + Pair.of(null, "v") + ) + }, + // last message without key. + { Arrays.asList( + Pair.of("k1", "v1"), + Pair.of(null, "v") + ) + }, + // two messages. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2") + ) + }, + // end with delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k1", "v2"), + Pair.of("k1", null) + ) + }, + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k2", null) + ) + }, + // the second scenario of end with delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k2", null), + Pair.of("k3", "v3"), + Pair.of("k3", null) + ) + }, + // the third scenario of end with delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k3", "v3"), + Pair.of("k2", null), + Pair.of("k3", null) + ) + }, + // all message delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k2", "v2"), + Pair.of("k3", "v3"), + Pair.of("k1", null), + Pair.of("k2", null), + Pair.of("k3", null) + ) + }, + // the second scenario of all message delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k1", null), + Pair.of("k2", "v2"), + Pair.of("k2", null), + Pair.of("k3", "v3"), + Pair.of("k3", null) + ) + }, + // the third scenario of all message delete by compaction. + { + Arrays.asList( + Pair.of("k1", null), + Pair.of("k2", null) + ) + }, + // the fourth scenario of all message delete by compaction. + { + Arrays.asList( + Pair.of("k1", "v1"), + Pair.of("k1", null) + ) + } + }; + } + + @Test(dataProvider = "messagesToSend") + public void testRaceConditionByCompactionAndGetLastMessageId(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(false, messagesToSend, 1, false); + } + + @Test(dataProvider = "messagesToSend") + public void testRaceConditionByCompactionAndGetLastBatchMessageId(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 1, false); + } + + @Test(dataProvider = "messagesToSend") + public void testRaceConditionByCompactionAndGetLastBatchMessageId2(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 3, false); + } + + @Test(dataProvider = "messagesToSend") + public void testReadMessageAfterCompaction(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(false, messagesToSend, 1, true); + } + + @Test(dataProvider = "messagesToSend") + public void testReadMessageAfterCompactionWithBatchFuture(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 1, true); + } + + @Test(dataProvider = "messagesToSend") + public void testReadMessageAfterCompactionWithBatchFuture2(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 3, true); + } + + /** + * Motivation: + * 1. If the last message with key `k` of a topic is null, the compactor will mark all messages for that key as + * deleted. At this time, the last message read compacted will be `{ml.lastConfirmPosition - 1}`. + * 2. When we call `getLastMessageId`, consumer will initialize the attribute`lastMessageIdInBroker` + * as `{ml.lastConfirmPosition}`, then when we call method `hasMessageAvailable` it's going to return + * `consumer.startMessageId < hasMessageAvailable`. + * From here we get that the last message for compactor and consumer is different, so there will be a situation + * where `hasMessageAvailable` returns `true` but can't read the message by read compacted. + */ + private void doTestRaceConditionByCompactionAndGetLastMessageId(boolean enabledBatch, + List> messagesToSend, + int sendMessagesLoopCount, + boolean compactionBeforeGetLastMessageId) + throws Exception { + cleanup(); + // Disable the scheduled task: compaction. + conf.setBrokerServiceCompactionMonitorIntervalInSeconds(Integer.MAX_VALUE); + // Disable the scheduled task: retention. + conf.setRetentionCheckIntervalInSeconds(Integer.MAX_VALUE); + setup(); + + String topicName = "persistent://my-property/use/my-ns/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Reader reader = pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .startMessageId(MessageId.earliest) + .receiverQueueSize(1) + .readCompacted(true) + .create(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(enabledBatch) + .create(); + + List> sendFutures = new ArrayList<>(); + for (int i = 0; i < sendMessagesLoopCount; i++) { + for (Pair messageToSend : messagesToSend) { + String key = messageToSend.getLeft(); + String value = messageToSend.getRight(); + if (key == null) { + sendFutures.add(producer.newMessage().value(value).sendAsync()); + } else { + sendFutures.add(producer.newMessage().key(key).value(value).sendAsync()); + } + } + producer.flush(); + } + FutureUtil.waitForAll(sendFutures).join(); + + // Trigger race condition of "compaction" and "getLastMessageId". + if (!compactionBeforeGetLastMessageId) { + reader.hasMessageAvailable(); + } + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + persistentTopic.triggerCompaction(); + if (compactionBeforeGetLastMessageId) { + reader.hasMessageAvailable(); + } + + Awaitility.await().untilAsserted(() -> { + PositionImpl lastConfirmPos = (PositionImpl) persistentTopic.getManagedLedger().getLastConfirmedEntry(); + PositionImpl markDeletePos = (PositionImpl) persistentTopic + .getSubscription(Compactor.COMPACTION_SUBSCRIPTION).getCursor().getMarkDeletedPosition(); + assertEquals(markDeletePos.getLedgerId(), lastConfirmPos.getLedgerId()); + assertEquals(markDeletePos.getEntryId(), lastConfirmPos.getEntryId()); + }); + + // Method "hasMessageAvailable" does not guarantee that a subsequent call to {@link #readNext()} will not + // block. But we can't always tell users there has messages and can not receive them. + int hasAvailableButReadNullTimes = 0; + while (reader.hasMessageAvailable()) { + Message message = reader.readNext(2, TimeUnit.SECONDS); + if (message == null) { + hasAvailableButReadNullTimes++; + } + assertTrue(hasAvailableButReadNullTimes < 10); + } + + // cleanup. + reader.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + @Test public void testSeekEarliestAfterCompaction() throws Exception { String topic = "persistent://my-property/use/my-ns/my-topic1"; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java new file mode 100644 index 0000000000000..aba55c960c01f --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java @@ -0,0 +1,434 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.compaction; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.service.Topic; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.BatchMessageIdImpl; +import org.apache.pulsar.client.impl.MessageIdImpl; +import org.apache.pulsar.client.impl.ReaderImpl; +import org.apache.pulsar.common.util.FutureUtil; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test(groups = "broker-impl") +public class GetLastMessageIdCompactedTest extends ProducerConsumerBase { + + @BeforeClass + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + } + + @AfterClass + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + // Disable the scheduled task: compaction. + conf.setBrokerServiceCompactionMonitorIntervalInSeconds(Integer.MAX_VALUE); + // Disable the scheduled task: retention. + conf.setRetentionCheckIntervalInSeconds(Integer.MAX_VALUE); + } + + private MessageIdImpl getLastMessageIdByTopic(String topicName) throws Exception{ + return (MessageIdImpl) pulsar.getBrokerService().getTopic(topicName, false) + .get().get().getLastMessageId().get(); + } + + private void triggerCompactionAndWait(String topicName) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + persistentTopic.triggerCompaction(); + Awaitility.await().untilAsserted(() -> { + PositionImpl lastConfirmPos = (PositionImpl) persistentTopic.getManagedLedger().getLastConfirmedEntry(); + PositionImpl markDeletePos = (PositionImpl) persistentTopic + .getSubscription(Compactor.COMPACTION_SUBSCRIPTION).getCursor().getMarkDeletedPosition(); + assertEquals(markDeletePos.getLedgerId(), lastConfirmPos.getLedgerId()); + assertEquals(markDeletePos.getEntryId(), lastConfirmPos.getEntryId()); + }); + } + + private void triggerLedgerSwitch(String topicName) throws Exception{ + admin.topics().unload(topicName); + Awaitility.await().until(() -> { + CompletableFuture> topicFuture = + pulsar.getBrokerService().getTopic(topicName, false); + if (!topicFuture.isDone() || topicFuture.isCompletedExceptionally()){ + return false; + } + Optional topicOptional = topicFuture.join(); + if (!topicOptional.isPresent()){ + return false; + } + PersistentTopic persistentTopic = (PersistentTopic) topicOptional.get(); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + return managedLedger.getState() == ManagedLedgerImpl.State.LedgerOpened; + }); + } + + private void clearAllTheLedgersOutdated(String topicName) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> { + CompletableFuture future = new CompletableFuture(); + managedLedger.trimConsumedLedgersInBackground(future); + future.join(); + return managedLedger.getLedgersInfo().size() == 1; + }); + } + + @Test + public void testGetLastMessageIdWhenLedgerEmpty() throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(messageId.getLedgerId(), -1); + assertEquals(messageId.getEntryId(), -1); + + // cleanup. + consumer.close(); + admin.topics().delete(topicName, false); + } + + @Test + public void testGetLastMessageIdWhenNoNonEmptyLedgerExists() throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + ReaderImpl reader = (ReaderImpl) pulsarClient.newReader(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .startMessageId(MessageId.earliest) + .readCompacted(false) + .create(); + + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .create(); + + producer.newMessage().key("k0").value("v0").sendAsync().get(); + reader.readNext(); + triggerLedgerSwitch(topicName); + clearAllTheLedgersOutdated(topicName); + + MessageIdImpl messageId = (MessageIdImpl) reader.getConsumer().getLastMessageId(); + assertEquals(messageId.getLedgerId(), -1); + assertEquals(messageId.getEntryId(), -1); + + // cleanup. + reader.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + + @DataProvider(name = "enabledBatch") + public Object[][] enabledBatch(){ + return new Object[][]{ + {true}, + {false} + }; + } + + @Test(dataProvider = "enabledBatch") + public void testGetLastMessageIdBeforeCompaction(boolean enabledBatch) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(enabledBatch) + .create(); + + List> sendFutures = new ArrayList<>(); + sendFutures.add(producer.newMessage().key("k0").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v2").sendAsync()); + producer.flush(); + sendFutures.add(producer.newMessage().key("k1").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v2").sendAsync()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + MessageIdImpl lastMessageIdExpected = getLastMessageIdByTopic(topicName); + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(lastMessageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); + assertEquals(lastMessageId.getEntryId(), lastMessageIdExpected.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) lastMessageIdExpected; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) lastMessageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); + } + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + + @Test(dataProvider = "enabledBatch") + public void testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(enabledBatch) + .create(); + + List> sendFutures = new ArrayList<>(); + sendFutures.add(producer.newMessage().key("k0").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v2").sendAsync()); + producer.flush(); + sendFutures.add(producer.newMessage().key("k1").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v2").sendAsync()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + triggerCompactionAndWait(topicName); + + MessageIdImpl lastMessageIdByTopic = getLastMessageIdByTopic(topicName); + MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(messageId.getLedgerId(), lastMessageIdByTopic.getLedgerId()); + assertEquals(messageId.getEntryId(), lastMessageIdByTopic.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) lastMessageIdByTopic; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); + } + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + + @Test(dataProvider = "enabledBatch") + public void testGetLastMessageIdAfterCompactionEndWithNullMsg(boolean enabledBatch) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + Producer producer; + if (enabledBatch){ + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(true) + .batchingMaxMessages(Integer.MAX_VALUE) + .batchingMaxBytes(Integer.MAX_VALUE) + .batchingMaxPublishDelay(2, TimeUnit.DAYS) + .create(); + } else { + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .create(); + } + + List> sendFutures = new ArrayList<>(); + sendFutures.add(producer.newMessage().key("k0").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v2").sendAsync()); + producer.flush(); + sendFutures.add(producer.newMessage().key("k1").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value(null).sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value(null).sendAsync()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + triggerCompactionAndWait(topicName); + + MessageIdImpl lastMessageIdExpected = (MessageIdImpl) sendFutures.get(2).get(); + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(lastMessageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); + assertEquals(lastMessageId.getEntryId(), lastMessageIdExpected.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdExpected = (BatchMessageIdImpl) lastMessageIdExpected; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) lastMessageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdExpected.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdExpected.getBatchIndex()); + } + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + + @Test(dataProvider = "enabledBatch") + public void testGetLastMessageIdAfterCompactionEndWithNullMsg2(boolean enabledBatch) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + Producer producer; + if (enabledBatch){ + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(true) + .batchingMaxMessages(Integer.MAX_VALUE) + .batchingMaxBytes(Integer.MAX_VALUE) + .batchingMaxPublishDelay(2, TimeUnit.DAYS) + .create(); + } else { + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .create(); + } + + List> sendFutures = new ArrayList<>(); + sendFutures.add(producer.newMessage().key("k0").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value("v1").sendAsync()); + producer.flush(); + sendFutures.add(producer.newMessage().key("k1").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value("v2").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value("v1").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value(null).sendAsync()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + triggerCompactionAndWait(topicName); + + MessageIdImpl lastMessageIdExpected = (MessageIdImpl) sendFutures.get(4).get(); + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(lastMessageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); + assertEquals(lastMessageId.getEntryId(), lastMessageIdExpected.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdExpected = (BatchMessageIdImpl) lastMessageIdExpected; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) lastMessageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdExpected.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), 5); + } + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } + + @Test(dataProvider = "enabledBatch") + public void testGetLastMessageIdAfterCompactionAllNullMsg(boolean enabledBatch) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + String subName = "sub"; + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topicName) + .subscriptionName(subName) + .receiverQueueSize(1) + .readCompacted(true) + .subscribe(); + Producer producer; + if (enabledBatch){ + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(true) + .batchingMaxMessages(Integer.MAX_VALUE) + .batchingMaxBytes(Integer.MAX_VALUE) + .batchingMaxPublishDelay(2, TimeUnit.DAYS) + .create(); + } else { + producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .enableBatching(false) + .create(); + } + + List> sendFutures = new ArrayList<>(); + sendFutures.add(producer.newMessage().key("k0").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k0").value(null).sendAsync()); + producer.flush(); + sendFutures.add(producer.newMessage().key("k1").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k1").value(null).sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value("v0").sendAsync()); + sendFutures.add(producer.newMessage().key("k2").value(null).sendAsync()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + triggerCompactionAndWait(topicName); + + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertFalse(lastMessageId instanceof BatchMessageIdImpl); + assertEquals(lastMessageId.getLedgerId(), -1); + assertEquals(lastMessageId.getEntryId(), -1); + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } +} diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java index 3fbab236a60ee..5f1753d99fdba 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Consumer.java @@ -545,6 +545,9 @@ CompletableFuture reconsumeLaterCumulativeAsync(Message message, /** * Get the last message id available for consume. + * Note tht in both cases below, the resulted message id of the second call will smaller than the first call. + * 1. Enabled read compacted, the last message has been deleted by compaction task. + * 2. If all the messages have been consumed and the all non-empty ledgers has been deleted. -1:-1 * * @return the last message id. */ diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java index 419a759f118ba..e0b790266d9c1 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Reader.java @@ -98,7 +98,7 @@ public interface Reader extends Closeable { * * 
{@code
      * while (reader.hasMessageAvailable()) {
-     *     Message msg = reader.readNext();
+     *     Message msg = reader.readNext(2, TimeUnit);
      *     // Do something
      * }
      *
@@ -107,7 +107,12 @@ public interface Reader extends Closeable {
      *
      * 
Note that this call might be blocking (see {@link #hasMessageAvailableAsync()} for async version) and
      * that even if this call returns true, that will not guarantee that a subsequent call to {@link #readNext()}
-     * will not block.
+     * will not block. Blocking occurs in the following scenarios (you can reproduce these scenarios by test
+     * ReproduceCantGetMessageButHasMessageAvailableTest)：
+     *   1. The last message has been deleted by compaction task.
+     *   2. Enabled compacted read and with batch sends, the last message read from compacted topic and marked
+     *   "compactedOut" has been lost by consumer.
+     *   3. No durable cursor exists, all messages deleted by trim ledgers task.
      *
      * @return true if the are messages available to be read, false otherwise
      * @throws PulsarClientException if there was any error in the operation
diff --git a/pulsar-proxy/tmp.1670482904047.properties b/pulsar-proxy/tmp.1670482904047.properties
new file mode 100644
index 0000000000000..49e06b815ef97
--- /dev/null
+++ b/pulsar-proxy/tmp.1670482904047.properties
@@ -0,0 +1 @@
+proxyAdditionalServlets=a,b,c