From 0c7b029177be293f9839b3b528131116088d29ef Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Tue, 6 Dec 2022 16:18:47 +0800 Subject: [PATCH 1/4] [improve] [test] Add test testTrimLedgerWillKeepsAtLeastOneLedgerWithData --- .../TopicTransactionBufferRecoverTest.java | 188 +++++++++++++++++- 1 file changed, 180 insertions(+), 8 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index 39c324d92f38c..6542571a86851 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -33,15 +33,19 @@ import java.lang.reflect.Field; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.NavigableMap; import java.util.Optional; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import lombok.AllArgsConstructor; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.bookkeeper.mledger.impl.ReadOnlyManagedLedgerImpl; @@ -74,6 +78,7 @@ import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.transaction.Transaction; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageIdImpl; @@ -119,18 +124,18 @@ protected void cleanup() throws Exception { } @DataProvider(name = "testTopic") - public Object[] testTopic() { - return new Object[] { - RECOVER_ABORT, - RECOVER_COMMIT + public Object[][] testTopic() { + return new Object[][] { + {RECOVER_ABORT}, + {RECOVER_COMMIT} }; } @DataProvider(name = "enableSnapshotSegment") - public Object[] testSnapshot() { - return new Boolean[] { - true, - false + public Object[][] testSnapshot() { + return new Object[][] { + {true}, + {false} }; } @@ -248,6 +253,173 @@ private void recoverTest(String testTopic) throws Exception { } + private ProducerAndConsumer makeManyTx(int txCount, String topicName, String subName) throws Exception { + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .subscriptionType(SubscriptionType.Shared) + .topic(topicName) + .isAckReceiptEnabled(true) + .acknowledgmentGroupTime(0, TimeUnit.SECONDS) + .subscriptionName(subName) + .subscribe(); + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topicName) + .sendTimeout(0, TimeUnit.SECONDS) + .enableBatching(false) + .batchingMaxMessages(2) + .create(); + producer.send("first message"); + boolean lastTxCommitted = false; + Message lastMessage = null; + for(int i = 0; i < txCount; i++) { + Transaction transaction = + pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().get(); + lastMessage = consumer.receive(); + producer.newMessage(transaction) + .value(new StringBuilder("tx message 0-") + .append(String.valueOf(lastMessage.getMessageId())).toString()).sendAsync(); + producer.newMessage(transaction) + .value(new StringBuilder("tx message 1-") + .append(String.valueOf(lastMessage.getMessageId())).toString()).sendAsync(); + consumer.acknowledgeAsync(lastMessage.getMessageId(), transaction); + if (i % 2 == 0) { + transaction.commit().get(); + lastTxCommitted = true; + } else { + transaction.abort().get(); + lastTxCommitted = false; + } + } + if (lastTxCommitted){ + Message msg = consumer.receive(); + consumer.acknowledge(msg); + } else { + consumer.acknowledge(lastMessage); + } + return new ProducerAndConsumer(producer, consumer); + } + + @AllArgsConstructor + private static class ProducerAndConsumer { + public Producer producer; + public Consumer consumer; + } + + private PersistentTopic findPersistentTopic(String topicName){ + for (PulsarService pulsarService : pulsarServiceList){ + CompletableFuture> future = pulsarService.getBrokerService().getTopic(topicName, false); + if (future == null || !future.isDone() || future.isCompletedExceptionally() || !future.join().isPresent()){ + continue; + } + return (PersistentTopic) future.join().get(); + } + throw new RuntimeException("topic[" + topicName + "] not found."); + } + + private void triggerSnapshot(String topicName){ + PersistentTopic persistentTopic = findPersistentTopic(topicName); + TopicTransactionBuffer topicTransactionBuffer = + (TopicTransactionBuffer) persistentTopic.getTransactionBuffer(); + topicTransactionBuffer.run(null); + } + + private void triggerLedgerTrims(String topicName){ + PersistentTopic persistentTopic = findPersistentTopic(topicName); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + CompletableFuture future = new CompletableFuture(); + managedLedger.trimConsumedLedgersInBackground(future); + future.join(); + } + + private Map getLedgers(String topicName){ + PersistentTopic persistentTopic = findPersistentTopic(topicName); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + return managedLedger.getLedgersInfo(); + } + + private void triggerCompact(String topicName) throws Exception { + PersistentTopic persistentTopic = findPersistentTopic(topicName); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + persistentTopic.getBrokerService().getPulsar().getCompactor().compact(topicName); + Awaitility.await().untilAsserted(() -> { + ManagedCursorImpl compaction = (ManagedCursorImpl) managedLedger.getCursors().get("__compaction"); + assertEquals(compaction.getMarkDeletedPosition().getLedgerId(), + managedLedger.getLastConfirmedEntry().getLedgerId()); + assertEquals(compaction.getMarkDeletedPosition().getEntryId(), + managedLedger.getLastConfirmedEntry().getEntryId()); + }); + ManagedCursorImpl compaction = (ManagedCursorImpl) managedLedger.getCursors().get("__compaction"); + log.info("===> cursor-compaction mark deleted position {}:{}", compaction.getMarkDeletedPosition().getLedgerId(), + compaction.getMarkDeletedPosition().getEntryId()); + } + + private void waitCursorDedup(String topicName) throws Exception { + PersistentTopic persistentTopic = findPersistentTopic(topicName); + ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); + persistentTopic.checkDeduplicationSnapshot(); + Awaitility.await().untilAsserted(() -> { + ManagedCursorImpl dedupCursor = (ManagedCursorImpl) managedLedger.getCursors().get("pulsar.dedup"); + assertEquals(dedupCursor.getMarkDeletedPosition().getLedgerId(), + managedLedger.getLastConfirmedEntry().getLedgerId()); + assertEquals(dedupCursor.getMarkDeletedPosition().getEntryId(), + managedLedger.getLastConfirmedEntry().getEntryId()); + }); + ManagedCursorImpl dedupCursor = (ManagedCursorImpl) managedLedger.getCursors().get("pulsar.dedup"); + log.info("===> cursor-dedup mark deleted position {}:{}", dedupCursor.getMarkDeletedPosition().getLedgerId(), + dedupCursor.getMarkDeletedPosition().getEntryId()); + } + + @Test + private void testTrimLedgerWillKeepsAtLeastOneLedgerWithData() throws Exception { + String topicName = String.format("persistent://%s/%s", NAMESPACE1, + "tx_recover_" + UUID.randomUUID().toString().replaceAll("-", "_")); + String subName = "sub"; + String transactionBufferTopicName = + String.format("persistent://%s/%s", NAMESPACE1, TRANSACTION_BUFFER_SNAPSHOT); + + // Make some data. + ProducerAndConsumer producerAndConsumer = null; + for (int i = 0; i < 5; i++) { + producerAndConsumer = makeManyTx(10, topicName, subName); + triggerSnapshot(topicName); + if (i != 4) { + // Do not close all clients. + producerAndConsumer.producer.close(); + producerAndConsumer.consumer.close(); + } + // Reload for create new ledger, and wait for topic reload. + admin.topics().unload(transactionBufferTopicName); + Awaitility.await().until(() -> { + try { + findPersistentTopic(transactionBufferTopicName); + return true; + } catch (Exception e) { + return false; + } + }); + } + + // Verify the last ledger will not be deleted. + Map ledgers = getLedgers(transactionBufferTopicName); + long lastLedgerHasData = -1; + for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledger : ledgers.values()){ + if (ledger.getEntries() > 0){ + lastLedgerHasData = Math.max(lastLedgerHasData, ledger.getLedgerId()); + } + } + log.info("===> ledgers before trim {}", ledgers.keySet()); + triggerCompact(transactionBufferTopicName); + waitCursorDedup(transactionBufferTopicName); + triggerLedgerTrims(transactionBufferTopicName); + ledgers = getLedgers(transactionBufferTopicName); + log.info("===> ledgers after trim {}", ledgers.keySet()); + assertTrue(ledgers.containsKey(lastLedgerHasData)); + + // cleanup. + producerAndConsumer.producer.close(); + producerAndConsumer.consumer.close(); + admin.topics().delete(topicName, false); + } + private void testTakeSnapshot() throws Exception { @Cleanup Producer producer = pulsarClient From b07a11dd6395b7f24244a455218f91cd71fcd140 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Wed, 7 Dec 2022 23:16:29 +0800 Subject: [PATCH 2/4] [fix] [broker] Fix reader can not get any messages but hasMessageAvailable always return true --- .../pulsar/compaction/CompactedTopicImpl.java | 21 +- .../pulsar/compaction/CompactionTest.java | 201 +++++++++++ .../GetLastMessageIdCompactedTest.java | 320 ++++++++++++++++++ 3 files changed, 540 insertions(+), 2 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java 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 c8114f9adb652..6a0d20e0f1dc7 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 @@ -104,12 +104,22 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor, || compactionHorizon.compareTo(cursorPosition) < 0) { cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, PositionImpl.LATEST); } else { + final PositionImpl mlLastConfirmPosition = + (PositionImpl) cursor.getManagedLedger().getLastConfirmedEntry(); compactedTopicContext.thenCompose( (context) -> findStartPoint(cursorPosition, context.ledger.getLastAddConfirmed(), context.cache) .thenCompose((startPoint) -> { - // do not need to read the compaction ledger if it is empty. - // the cursor just needs to be set to the compaction horizon if (startPoint == COMPACT_LEDGER_EMPTY) { + // All messages have deleted by compaction, just read the last message from original + // cursor. + if (compactionHorizon.compareTo(mlLastConfirmPosition) >= 0) { + cursor.seek(compactionHorizon); + cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, + PositionImpl.LATEST); + return CompletableFuture.completedFuture(null); + } + // do not need to read the compaction ledger if it is empty. + // the cursor just needs to be set to the compaction horizon cursor.seek(compactionHorizon.getNext()); callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); return CompletableFuture.completedFuture(null); @@ -121,6 +131,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor, } else { long endPoint = Math.min(context.ledger.getLastAddConfirmed(), startPoint + numberOfEntriesToRead); + if (startPoint == context.ledger.getLastAddConfirmed() + && compactionHorizon.compareTo(mlLastConfirmPosition) >= 0){ + cursor.seek(compactionHorizon); + cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, + PositionImpl.LATEST); + return CompletableFuture.completedFuture(null); + } if (startPoint == NEWER_THAN_COMPACTED) { cursor.seek(compactionHorizon.getNext()); callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); 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..36945629cd7b3 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,203 @@ 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); + } + + @Test(dataProvider = "messagesToSend") + public void testRaceConditionByCompactionAndGetLastBatchMessageId(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 1); + } + + @Test(dataProvider = "messagesToSend") + public void testRaceConditionByCompactionAndGetLastBatchMessageId2(List> messagesToSend) + throws Exception { + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 3); + } + + /** + * 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) + 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". + reader.hasMessageAvailable(); + 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()); + }); + + // 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..15898c0d18db6 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java @@ -0,0 +1,320 @@ +/* + * 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.mockito.Mockito.spy; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.BrokerTestUtil; +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.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 void sendManyBatchedMessages(int msgCountPerEntry, int entryCount, String topicName) + throws Exception { + sendManyBatchedMessages(msgCountPerEntry, entryCount, topicName, "1"); + } + + private MessageIdImpl getLastMessageIdByTopic(String topicName) throws Exception{ + return (MessageIdImpl) pulsar.getBrokerService().getTopic(topicName, false) + .get().get().getLastMessageId().get(); + } + + private void sendManyBatchedMessages(int msgCountPerEntry, int entryCount, String topicName, String key) + throws Exception { + Producer producer = pulsarClient.newProducer(Schema.JSON(String.class)) + .topic(topicName) + .enableBatching(true) + .batchingMaxPublishDelay(Integer.MAX_VALUE, TimeUnit.SECONDS) + .batchingMaxMessages(Integer.MAX_VALUE) + .create(); + for (int i = 0; i < entryCount; i++){ + for (int j = 0; j < msgCountPerEntry; j++){ + producer.newMessage().key(key).value(String.format("entry-seq[%s], batch_index[%s]", i, j)).sendAsync(); + } + producer.flush(); + } + producer.close(); + } + + 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()); + }); + } + + @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); + } + + @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 lastMessageIdByTopic = getLastMessageIdByTopic(topicName); + MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(messageId.getLedgerId(), lastMessageIdByTopic.getLedgerId()); + assertEquals(messageId.getEntryId(), lastMessageIdByTopic.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) getLastMessageIdByTopic(topicName); + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) consumer.getLastMessageId(); + 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 testGetLastMessageIdBeforeCompactionEndWithNullMsg(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(); + // TODO 这个问题解不了。明天开会说下吧。 + 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()); + producer.flush(); + FutureUtil.waitForAll(sendFutures).join(); + + MessageIdImpl lastMessageIdExpected = (MessageIdImpl) sendFutures.get(2).get(); + MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(messageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); + assertEquals(messageId.getEntryId(), lastMessageIdExpected.getEntryId()); + if (enabledBatch){ + BatchMessageIdImpl lastBatchMessageIdExpected = (BatchMessageIdImpl) getLastMessageIdByTopic(topicName); + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) consumer.getLastMessageId(); + 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 testGetLastMessageIdBeforeCompactionAllNullMsg(boolean enabledBatch) throws Exception { + } + + @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) consumer.getLastMessageId(); + 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 testGetLastMessageIdAfterCompactionAndEndWithNullMsg(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(null).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()); + 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) consumer.getLastMessageId(); + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); + } + + // cleanup. + consumer.close(); + producer.close(); + admin.topics().delete(topicName, false); + } +} From eb7912e74b5121bb1efc1825d254ba2d8b205b35 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 8 Dec 2022 03:46:47 +0800 Subject: [PATCH 3/4] [fix] [broker] Fix reader can not get any messages but hasMessageAvailable always return true --- .../TopicTransactionBufferRecoverTest.java | 6 +- .../pulsar/compaction/CompactionTest.java | 34 +++- .../GetLastMessageIdCompactedTest.java | 181 ++++++++++++------ 3 files changed, 156 insertions(+), 65 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index 6542571a86851..8193cbde36a1c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -369,7 +369,7 @@ private void waitCursorDedup(String topicName) throws Exception { } @Test - private void testTrimLedgerWillKeepsAtLeastOneLedgerWithData() throws Exception { + private void testLastNonEmptyLedgerWillNotBeDeleteByTrimLedgers() throws Exception { String topicName = String.format("persistent://%s/%s", NAMESPACE1, "tx_recover_" + UUID.randomUUID().toString().replaceAll("-", "_")); String subName = "sub"; @@ -401,8 +401,8 @@ private void testTrimLedgerWillKeepsAtLeastOneLedgerWithData() throws Exception // Verify the last ledger will not be deleted. Map ledgers = getLedgers(transactionBufferTopicName); long lastLedgerHasData = -1; - for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledger : ledgers.values()){ - if (ledger.getEntries() > 0){ + for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledger : ledgers.values()) { + if (ledger.getEntries() > 0) { lastLedgerHasData = Math.max(lastLedgerHasData, ledger.getLedgerId()); } } 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 36945629cd7b3..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 @@ -426,19 +426,37 @@ public Object[][] messagesToSend() { @Test(dataProvider = "messagesToSend") public void testRaceConditionByCompactionAndGetLastMessageId(List> messagesToSend) throws Exception { - doTestRaceConditionByCompactionAndGetLastMessageId(false, messagesToSend, 1); + doTestRaceConditionByCompactionAndGetLastMessageId(false, messagesToSend, 1, false); } @Test(dataProvider = "messagesToSend") public void testRaceConditionByCompactionAndGetLastBatchMessageId(List> messagesToSend) throws Exception { - doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 1); + doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 1, false); } @Test(dataProvider = "messagesToSend") public void testRaceConditionByCompactionAndGetLastBatchMessageId2(List> messagesToSend) throws Exception { - doTestRaceConditionByCompactionAndGetLastMessageId(true, messagesToSend, 3); + 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); } /** @@ -453,7 +471,8 @@ public void testRaceConditionByCompactionAndGetLastBatchMessageId2(List> messagesToSend, - int sendMessagesLoopCount) + int sendMessagesLoopCount, + boolean compactionBeforeGetLastMessageId) throws Exception { cleanup(); // Disable the scheduled task: compaction. @@ -492,10 +511,15 @@ private void doTestRaceConditionByCompactionAndGetLastMessageId(boolean enabledB FutureUtil.waitForAll(sendFutures).join(); // Trigger race condition of "compaction" and "getLastMessageId". - reader.hasMessageAvailable(); + 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(); 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 index 15898c0d18db6..4fc6b2775e5d8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java @@ -18,12 +18,8 @@ */ package org.apache.pulsar.compaction; -import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; -import static org.testng.Assert.assertNotNull; -import static org.testng.Assert.assertNull; -import static org.testng.Assert.assertTrue; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -70,11 +66,6 @@ protected void doInitConf() throws Exception { conf.setRetentionCheckIntervalInSeconds(Integer.MAX_VALUE); } - private void sendManyBatchedMessages(int msgCountPerEntry, int entryCount, String topicName) - throws Exception { - sendManyBatchedMessages(msgCountPerEntry, entryCount, topicName, "1"); - } - private MessageIdImpl getLastMessageIdByTopic(String topicName) throws Exception{ return (MessageIdImpl) pulsar.getBrokerService().getTopic(topicName, false) .get().get().getLastMessageId().get(); @@ -163,13 +154,13 @@ public void testGetLastMessageIdBeforeCompaction(boolean enabledBatch) throws Ex producer.flush(); FutureUtil.waitForAll(sendFutures).join(); - MessageIdImpl lastMessageIdByTopic = getLastMessageIdByTopic(topicName); - MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); - assertEquals(messageId.getLedgerId(), lastMessageIdByTopic.getLedgerId()); - assertEquals(messageId.getEntryId(), lastMessageIdByTopic.getEntryId()); + MessageIdImpl lastMessageIdExpected = getLastMessageIdByTopic(topicName); + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertEquals(lastMessageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); + assertEquals(lastMessageId.getEntryId(), lastMessageIdExpected.getEntryId()); if (enabledBatch){ - BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) getLastMessageIdByTopic(topicName); - BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) consumer.getLastMessageId(); + BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) lastMessageIdExpected; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) lastMessageId; assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); } @@ -180,8 +171,10 @@ public void testGetLastMessageIdBeforeCompaction(boolean enabledBatch) throws Ex admin.topics().delete(topicName, false); } + + @Test(dataProvider = "enabledBatch") - public void testGetLastMessageIdBeforeCompactionEndWithNullMsg(boolean enabledBatch) throws Exception { + public void testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); String subName = "sub"; Consumer consumer = pulsarClient.newConsumer(Schema.STRING) @@ -200,22 +193,23 @@ public void testGetLastMessageIdBeforeCompactionEndWithNullMsg(boolean enabledBa sendFutures.add(producer.newMessage().key("k0").value("v1").sendAsync()); sendFutures.add(producer.newMessage().key("k0").value("v2").sendAsync()); producer.flush(); - // TODO 这个问题解不了。明天开会说下吧。 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("k1").value("v2").sendAsync()); producer.flush(); FutureUtil.waitForAll(sendFutures).join(); - MessageIdImpl lastMessageIdExpected = (MessageIdImpl) sendFutures.get(2).get(); + triggerCompactionAndWait(topicName); + + MessageIdImpl lastMessageIdByTopic = getLastMessageIdByTopic(topicName); MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); - assertEquals(messageId.getLedgerId(), lastMessageIdExpected.getLedgerId()); - assertEquals(messageId.getEntryId(), lastMessageIdExpected.getEntryId()); + assertEquals(messageId.getLedgerId(), lastMessageIdByTopic.getLedgerId()); + assertEquals(messageId.getEntryId(), lastMessageIdByTopic.getEntryId()); if (enabledBatch){ - BatchMessageIdImpl lastBatchMessageIdExpected = (BatchMessageIdImpl) getLastMessageIdByTopic(topicName); - BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) consumer.getLastMessageId(); - assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdExpected.getBatchSize()); - assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdExpected.getBatchIndex()); + BatchMessageIdImpl lastBatchMessageIdByTopic = (BatchMessageIdImpl) lastMessageIdByTopic; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); } // cleanup. @@ -225,11 +219,66 @@ public void testGetLastMessageIdBeforeCompactionEndWithNullMsg(boolean enabledBa } @Test(dataProvider = "enabledBatch") - public void testGetLastMessageIdBeforeCompactionAllNullMsg(boolean enabledBatch) throws Exception { + 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 testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exception { + public void testGetLastMessageIdAfterCompactionEndWithNullMsg2(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); String subName = "sub"; Consumer consumer = pulsarClient.newConsumer(Schema.STRING) @@ -238,33 +287,46 @@ public void testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exc .receiverQueueSize(1) .readCompacted(true) .subscribe(); - Producer producer = pulsarClient.newProducer(Schema.STRING) - .topic(topicName) - .enableBatching(enabledBatch) - .create(); + 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("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 lastMessageIdByTopic = getLastMessageIdByTopic(topicName); - MessageIdImpl messageId = (MessageIdImpl) consumer.getLastMessageId(); - assertEquals(messageId.getLedgerId(), lastMessageIdByTopic.getLedgerId()); - assertEquals(messageId.getEntryId(), lastMessageIdByTopic.getEntryId()); + 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 lastBatchMessageIdByTopic = (BatchMessageIdImpl) lastMessageIdByTopic; - BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) consumer.getLastMessageId(); - assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); - assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); + BatchMessageIdImpl lastBatchMessageIdExpected = (BatchMessageIdImpl) lastMessageIdExpected; + BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) lastMessageId; + assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdExpected.getBatchSize()); + assertEquals(batchMessageId.getBatchIndex(), 5); } // cleanup. @@ -274,7 +336,7 @@ public void testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exc } @Test(dataProvider = "enabledBatch") - public void testGetLastMessageIdAfterCompactionAndEndWithNullMsg(boolean enabledBatch) throws Exception { + public void testGetLastMessageIdAfterCompactionAllNullMsg(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); String subName = "sub"; Consumer consumer = pulsarClient.newConsumer(Schema.STRING) @@ -283,34 +345,39 @@ public void testGetLastMessageIdAfterCompactionAndEndWithNullMsg(boolean enabled .receiverQueueSize(1) .readCompacted(true) .subscribe(); - Producer producer = pulsarClient.newProducer(Schema.STRING) - .topic(topicName) - .enableBatching(enabledBatch) - .create(); + 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(null).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(null).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) consumer.getLastMessageId(); - assertEquals(batchMessageId.getBatchSize(), lastBatchMessageIdByTopic.getBatchSize()); - assertEquals(batchMessageId.getBatchIndex(), lastBatchMessageIdByTopic.getBatchIndex()); - } + MessageIdImpl lastMessageId = (MessageIdImpl) consumer.getLastMessageId(); + assertFalse(lastMessageId instanceof BatchMessageIdImpl); + assertEquals(lastMessageId.getLedgerId(), -1); + assertEquals(lastMessageId.getEntryId(), -1); // cleanup. consumer.close(); From 805b8ab0aee85e8131063765452b3d5fb8749253 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Fri, 9 Dec 2022 14:34:57 +0800 Subject: [PATCH 4/4] revert and add test --- .../mledger/impl/ManagedLedgerImpl.java | 2 +- ...SingleSnapshotAbortedTxnProcessorImpl.java | 1 + .../pulsar/compaction/CompactedTopicImpl.java | 21 +- .../TopicTransactionBufferRecoverTest.java | 188 +------- ...tGetMessageButHasMessageAvailableTest.java | 439 ++++++++++++++++++ .../GetLastMessageIdCompactedTest.java | 85 +++- .../apache/pulsar/client/api/Consumer.java | 3 + .../org/apache/pulsar/client/api/Reader.java | 9 +- pulsar-proxy/tmp.1670482904047.properties | 1 + 9 files changed, 528 insertions(+), 221 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/client/api/ReproduceCantGetMessageButHasMessageAvailableTest.java create mode 100644 pulsar-proxy/tmp.1670482904047.properties 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/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java index 6a0d20e0f1dc7..c8114f9adb652 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 @@ -104,22 +104,12 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor, || compactionHorizon.compareTo(cursorPosition) < 0) { cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, PositionImpl.LATEST); } else { - final PositionImpl mlLastConfirmPosition = - (PositionImpl) cursor.getManagedLedger().getLastConfirmedEntry(); compactedTopicContext.thenCompose( (context) -> findStartPoint(cursorPosition, context.ledger.getLastAddConfirmed(), context.cache) .thenCompose((startPoint) -> { + // do not need to read the compaction ledger if it is empty. + // the cursor just needs to be set to the compaction horizon if (startPoint == COMPACT_LEDGER_EMPTY) { - // All messages have deleted by compaction, just read the last message from original - // cursor. - if (compactionHorizon.compareTo(mlLastConfirmPosition) >= 0) { - cursor.seek(compactionHorizon); - cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, - PositionImpl.LATEST); - return CompletableFuture.completedFuture(null); - } - // do not need to read the compaction ledger if it is empty. - // the cursor just needs to be set to the compaction horizon cursor.seek(compactionHorizon.getNext()); callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); return CompletableFuture.completedFuture(null); @@ -131,13 +121,6 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor, } else { long endPoint = Math.min(context.ledger.getLastAddConfirmed(), startPoint + numberOfEntriesToRead); - if (startPoint == context.ledger.getLastAddConfirmed() - && compactionHorizon.compareTo(mlLastConfirmPosition) >= 0){ - cursor.seek(compactionHorizon); - cursor.asyncReadEntriesOrWait(numberOfEntriesToRead, callback, readEntriesCtx, - PositionImpl.LATEST); - return CompletableFuture.completedFuture(null); - } if (startPoint == NEWER_THAN_COMPACTED) { cursor.seek(compactionHorizon.getNext()); callback.readEntriesComplete(Collections.emptyList(), readEntriesCtx); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java index 8193cbde36a1c..39c324d92f38c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java @@ -33,19 +33,15 @@ import java.lang.reflect.Field; import java.util.LinkedList; import java.util.List; -import java.util.Map; import java.util.NavigableMap; import java.util.Optional; -import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import lombok.AllArgsConstructor; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; -import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.bookkeeper.mledger.impl.ReadOnlyManagedLedgerImpl; @@ -78,7 +74,6 @@ import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.transaction.Transaction; import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.client.impl.MessageIdImpl; @@ -124,18 +119,18 @@ protected void cleanup() throws Exception { } @DataProvider(name = "testTopic") - public Object[][] testTopic() { - return new Object[][] { - {RECOVER_ABORT}, - {RECOVER_COMMIT} + public Object[] testTopic() { + return new Object[] { + RECOVER_ABORT, + RECOVER_COMMIT }; } @DataProvider(name = "enableSnapshotSegment") - public Object[][] testSnapshot() { - return new Object[][] { - {true}, - {false} + public Object[] testSnapshot() { + return new Boolean[] { + true, + false }; } @@ -253,173 +248,6 @@ private void recoverTest(String testTopic) throws Exception { } - private ProducerAndConsumer makeManyTx(int txCount, String topicName, String subName) throws Exception { - Consumer consumer = pulsarClient.newConsumer(Schema.STRING) - .subscriptionType(SubscriptionType.Shared) - .topic(topicName) - .isAckReceiptEnabled(true) - .acknowledgmentGroupTime(0, TimeUnit.SECONDS) - .subscriptionName(subName) - .subscribe(); - Producer producer = pulsarClient.newProducer(Schema.STRING) - .topic(topicName) - .sendTimeout(0, TimeUnit.SECONDS) - .enableBatching(false) - .batchingMaxMessages(2) - .create(); - producer.send("first message"); - boolean lastTxCommitted = false; - Message lastMessage = null; - for(int i = 0; i < txCount; i++) { - Transaction transaction = - pulsarClient.newTransaction().withTransactionTimeout(10, TimeUnit.SECONDS).build().get(); - lastMessage = consumer.receive(); - producer.newMessage(transaction) - .value(new StringBuilder("tx message 0-") - .append(String.valueOf(lastMessage.getMessageId())).toString()).sendAsync(); - producer.newMessage(transaction) - .value(new StringBuilder("tx message 1-") - .append(String.valueOf(lastMessage.getMessageId())).toString()).sendAsync(); - consumer.acknowledgeAsync(lastMessage.getMessageId(), transaction); - if (i % 2 == 0) { - transaction.commit().get(); - lastTxCommitted = true; - } else { - transaction.abort().get(); - lastTxCommitted = false; - } - } - if (lastTxCommitted){ - Message msg = consumer.receive(); - consumer.acknowledge(msg); - } else { - consumer.acknowledge(lastMessage); - } - return new ProducerAndConsumer(producer, consumer); - } - - @AllArgsConstructor - private static class ProducerAndConsumer { - public Producer producer; - public Consumer consumer; - } - - private PersistentTopic findPersistentTopic(String topicName){ - for (PulsarService pulsarService : pulsarServiceList){ - CompletableFuture> future = pulsarService.getBrokerService().getTopic(topicName, false); - if (future == null || !future.isDone() || future.isCompletedExceptionally() || !future.join().isPresent()){ - continue; - } - return (PersistentTopic) future.join().get(); - } - throw new RuntimeException("topic[" + topicName + "] not found."); - } - - private void triggerSnapshot(String topicName){ - PersistentTopic persistentTopic = findPersistentTopic(topicName); - TopicTransactionBuffer topicTransactionBuffer = - (TopicTransactionBuffer) persistentTopic.getTransactionBuffer(); - topicTransactionBuffer.run(null); - } - - private void triggerLedgerTrims(String topicName){ - PersistentTopic persistentTopic = findPersistentTopic(topicName); - ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); - CompletableFuture future = new CompletableFuture(); - managedLedger.trimConsumedLedgersInBackground(future); - future.join(); - } - - private Map getLedgers(String topicName){ - PersistentTopic persistentTopic = findPersistentTopic(topicName); - ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); - return managedLedger.getLedgersInfo(); - } - - private void triggerCompact(String topicName) throws Exception { - PersistentTopic persistentTopic = findPersistentTopic(topicName); - ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); - persistentTopic.getBrokerService().getPulsar().getCompactor().compact(topicName); - Awaitility.await().untilAsserted(() -> { - ManagedCursorImpl compaction = (ManagedCursorImpl) managedLedger.getCursors().get("__compaction"); - assertEquals(compaction.getMarkDeletedPosition().getLedgerId(), - managedLedger.getLastConfirmedEntry().getLedgerId()); - assertEquals(compaction.getMarkDeletedPosition().getEntryId(), - managedLedger.getLastConfirmedEntry().getEntryId()); - }); - ManagedCursorImpl compaction = (ManagedCursorImpl) managedLedger.getCursors().get("__compaction"); - log.info("===> cursor-compaction mark deleted position {}:{}", compaction.getMarkDeletedPosition().getLedgerId(), - compaction.getMarkDeletedPosition().getEntryId()); - } - - private void waitCursorDedup(String topicName) throws Exception { - PersistentTopic persistentTopic = findPersistentTopic(topicName); - ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); - persistentTopic.checkDeduplicationSnapshot(); - Awaitility.await().untilAsserted(() -> { - ManagedCursorImpl dedupCursor = (ManagedCursorImpl) managedLedger.getCursors().get("pulsar.dedup"); - assertEquals(dedupCursor.getMarkDeletedPosition().getLedgerId(), - managedLedger.getLastConfirmedEntry().getLedgerId()); - assertEquals(dedupCursor.getMarkDeletedPosition().getEntryId(), - managedLedger.getLastConfirmedEntry().getEntryId()); - }); - ManagedCursorImpl dedupCursor = (ManagedCursorImpl) managedLedger.getCursors().get("pulsar.dedup"); - log.info("===> cursor-dedup mark deleted position {}:{}", dedupCursor.getMarkDeletedPosition().getLedgerId(), - dedupCursor.getMarkDeletedPosition().getEntryId()); - } - - @Test - private void testLastNonEmptyLedgerWillNotBeDeleteByTrimLedgers() throws Exception { - String topicName = String.format("persistent://%s/%s", NAMESPACE1, - "tx_recover_" + UUID.randomUUID().toString().replaceAll("-", "_")); - String subName = "sub"; - String transactionBufferTopicName = - String.format("persistent://%s/%s", NAMESPACE1, TRANSACTION_BUFFER_SNAPSHOT); - - // Make some data. - ProducerAndConsumer producerAndConsumer = null; - for (int i = 0; i < 5; i++) { - producerAndConsumer = makeManyTx(10, topicName, subName); - triggerSnapshot(topicName); - if (i != 4) { - // Do not close all clients. - producerAndConsumer.producer.close(); - producerAndConsumer.consumer.close(); - } - // Reload for create new ledger, and wait for topic reload. - admin.topics().unload(transactionBufferTopicName); - Awaitility.await().until(() -> { - try { - findPersistentTopic(transactionBufferTopicName); - return true; - } catch (Exception e) { - return false; - } - }); - } - - // Verify the last ledger will not be deleted. - Map ledgers = getLedgers(transactionBufferTopicName); - long lastLedgerHasData = -1; - for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ledger : ledgers.values()) { - if (ledger.getEntries() > 0) { - lastLedgerHasData = Math.max(lastLedgerHasData, ledger.getLedgerId()); - } - } - log.info("===> ledgers before trim {}", ledgers.keySet()); - triggerCompact(transactionBufferTopicName); - waitCursorDedup(transactionBufferTopicName); - triggerLedgerTrims(transactionBufferTopicName); - ledgers = getLedgers(transactionBufferTopicName); - log.info("===> ledgers after trim {}", ledgers.keySet()); - assertTrue(ledgers.containsKey(lastLedgerHasData)); - - // cleanup. - producerAndConsumer.producer.close(); - producerAndConsumer.consumer.close(); - admin.topics().delete(topicName, false); - } - private void testTakeSnapshot() throws Exception { @Cleanup Producer producer = pulsarClient 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/GetLastMessageIdCompactedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java index 4fc6b2775e5d8..aba55c960c01f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java @@ -22,10 +22,13 @@ 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; @@ -34,6 +37,7 @@ 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; @@ -71,23 +75,6 @@ private MessageIdImpl getLastMessageIdByTopic(String topicName) throws Exception .get().get().getLastMessageId().get(); } - private void sendManyBatchedMessages(int msgCountPerEntry, int entryCount, String topicName, String key) - throws Exception { - Producer producer = pulsarClient.newProducer(Schema.JSON(String.class)) - .topic(topicName) - .enableBatching(true) - .batchingMaxPublishDelay(Integer.MAX_VALUE, TimeUnit.SECONDS) - .batchingMaxMessages(Integer.MAX_VALUE) - .create(); - for (int i = 0; i < entryCount; i++){ - for (int j = 0; j < msgCountPerEntry; j++){ - producer.newMessage().key(key).value(String.format("entry-seq[%s], batch_index[%s]", i, j)).sendAsync(); - } - producer.flush(); - } - producer.close(); - } - private void triggerCompactionAndWait(String topicName) throws Exception { PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); @@ -101,6 +88,36 @@ private void triggerCompactionAndWait(String topicName) throws Exception { }); } + 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"); @@ -120,6 +137,38 @@ public void testGetLastMessageIdWhenLedgerEmpty() throws Exception { 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[][]{ @@ -171,8 +220,6 @@ public void testGetLastMessageIdBeforeCompaction(boolean enabledBatch) throws Ex admin.topics().delete(topicName, false); } - - @Test(dataProvider = "enabledBatch") public void testGetLastMessageIdAfterCompaction(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); 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