| /* |
| * 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.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION; |
| import com.github.benmanes.caffeine.cache.AsyncLoadingCache; |
| import com.google.common.collect.Sets; |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.Unpooled; |
| import java.nio.charset.Charset; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.Random; |
| import java.util.UUID; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.stream.IntStream; |
| import lombok.Cleanup; |
| import org.apache.bookkeeper.client.BKException; |
| import org.apache.bookkeeper.client.BookKeeper; |
| import org.apache.bookkeeper.client.LedgerHandle; |
| import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; |
| import org.apache.bookkeeper.mledger.impl.PositionImpl; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.commons.lang3.tuple.Triple; |
| import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; |
| import org.apache.pulsar.broker.service.Topic; |
| import org.apache.pulsar.broker.service.persistent.PersistentTopic; |
| import org.apache.pulsar.client.admin.LongRunningProcessStatus; |
| import org.apache.pulsar.client.admin.PulsarAdminException; |
| import org.apache.pulsar.client.api.Message; |
| import org.apache.pulsar.client.api.MessageId; |
| import org.apache.pulsar.client.api.Producer; |
| import org.apache.pulsar.client.api.ProducerBuilder; |
| import org.apache.pulsar.client.api.RawMessage; |
| import org.apache.pulsar.client.api.Reader; |
| import org.apache.pulsar.client.api.Schema; |
| import org.apache.pulsar.client.impl.RawMessageImpl; |
| import org.apache.pulsar.client.impl.ReaderImpl; |
| import org.apache.pulsar.common.api.proto.MessageIdData; |
| import org.apache.pulsar.common.policies.data.ClusterData; |
| import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; |
| import org.apache.pulsar.common.policies.data.TenantInfoImpl; |
| import org.apache.pulsar.common.util.FutureUtil; |
| import org.awaitility.Awaitility; |
| import org.testng.Assert; |
| import org.testng.annotations.AfterMethod; |
| import org.testng.annotations.BeforeMethod; |
| import org.testng.annotations.DataProvider; |
| import org.testng.annotations.Test; |
| |
| @Test(groups = "broker-compaction") |
| public class CompactedTopicTest extends MockedPulsarServiceBaseTest { |
| private final Random r = new Random(0); |
| |
| @DataProvider(name = "batchEnabledProvider") |
| public Object[][] batchEnabledProvider() { |
| return new Object[][] { |
| { Boolean.FALSE }, |
| { Boolean.TRUE } |
| }; |
| } |
| |
| @BeforeMethod |
| @Override |
| public void setup() throws Exception { |
| super.internalSetup(); |
| |
| admin.clusters().createCluster("use", |
| ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); |
| admin.tenants().createTenant("my-property", |
| new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("use"))); |
| admin.namespaces().createNamespace("my-property/use/my-ns"); |
| } |
| |
| @AfterMethod(alwaysRun = true) |
| @Override |
| public void cleanup() throws Exception { |
| super.internalCleanup(); |
| } |
| |
| /** |
| * Build a compacted ledger, and return the id of the ledger, the position of the different |
| * entries in the ledger, and a list of gaps, and the entry which should be returned after the gap. |
| */ |
| private Triple<Long, List<Pair<MessageIdData,Long>>, List<Pair<MessageIdData,Long>>> |
| buildCompactedLedger(BookKeeper bk, int count) |
| throws Exception { |
| LedgerHandle lh = bk.createLedger(1, 1, |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD); |
| List<Pair<MessageIdData,Long>> positions = new ArrayList<>(); |
| List<Pair<MessageIdData,Long>> idsInGaps = new ArrayList<>(); |
| |
| AtomicLong ledgerIds = new AtomicLong(10L); |
| AtomicLong entryIds = new AtomicLong(0L); |
| CompletableFuture.allOf( |
| IntStream.range(0, count).mapToObj((i) -> { |
| List<MessageIdData> idsInGap = new ArrayList<>(); |
| if (r.nextInt(10) == 1) { |
| long delta = r.nextInt(10) + 1; |
| idsInGap.add(new MessageIdData() |
| .setLedgerId(ledgerIds.get()) |
| .setEntryId(entryIds.get() + 1)); |
| ledgerIds.addAndGet(delta); |
| entryIds.set(0); |
| } |
| long delta = r.nextInt(5); |
| if (delta != 0) { |
| idsInGap.add(new MessageIdData() |
| .setLedgerId(ledgerIds.get()) |
| .setEntryId(entryIds.get() + 1)); |
| } |
| MessageIdData id = new MessageIdData() |
| .setLedgerId(ledgerIds.get()) |
| .setEntryId(entryIds.addAndGet(delta + 1)); |
| |
| @Cleanup |
| RawMessage m = new RawMessageImpl(id, Unpooled.EMPTY_BUFFER); |
| |
| CompletableFuture<Void> f = new CompletableFuture<>(); |
| ByteBuf buffer = m.serialize(); |
| |
| lh.asyncAddEntry(buffer, |
| (rc, ledger, eid, ctx) -> { |
| if (rc != BKException.Code.OK) { |
| f.completeExceptionally(BKException.create(rc)); |
| } else { |
| positions.add(Pair.of(id, eid)); |
| idsInGap.forEach((gid) -> idsInGaps.add(Pair.of(gid, eid))); |
| f.complete(null); |
| } |
| }, null); |
| return f; |
| }).toArray(CompletableFuture[]::new)).get(); |
| lh.close(); |
| |
| return Triple.of(lh.getId(), positions, idsInGaps); |
| } |
| |
| @Test |
| public void testEntryLookup() throws Exception { |
| BookKeeper bk = pulsar.getBookKeeperClientFactory().create( |
| this.conf, null, null, Optional.empty(), null); |
| |
| Triple<Long, List<Pair<MessageIdData, Long>>, List<Pair<MessageIdData, Long>>> compactedLedgerData |
| = buildCompactedLedger(bk, 500); |
| |
| List<Pair<MessageIdData, Long>> positions = compactedLedgerData.getMiddle(); |
| List<Pair<MessageIdData, Long>> idsInGaps = compactedLedgerData.getRight(); |
| |
| LedgerHandle lh = bk.openLedger(compactedLedgerData.getLeft(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD); |
| long lastEntryId = lh.getLastAddConfirmed(); |
| AsyncLoadingCache<Long,MessageIdData> cache = CompactedTopicImpl.createCache(lh, 50); |
| |
| MessageIdData firstPositionId = positions.get(0).getLeft(); |
| Pair<MessageIdData, Long> lastPosition = positions.get(positions.size() - 1); |
| |
| // check ids before and after ids in compacted ledger |
| Assert.assertEquals(CompactedTopicImpl.findStartPoint(new PositionImpl(0, 0), lastEntryId, cache).get(), |
| Long.valueOf(0)); |
| Assert.assertEquals(CompactedTopicImpl.findStartPoint(new PositionImpl(Long.MAX_VALUE, 0), |
| lastEntryId, cache).get(), |
| Long.valueOf(CompactedTopicImpl.NEWER_THAN_COMPACTED)); |
| |
| // entry 0 is never in compacted ledger due to how we generate dummy |
| Assert.assertEquals(CompactedTopicImpl.findStartPoint(new PositionImpl(firstPositionId.getLedgerId(), 0), |
| lastEntryId, cache).get(), |
| Long.valueOf(0)); |
| // check next id after last id in compacted ledger |
| Assert.assertEquals(CompactedTopicImpl.findStartPoint(new PositionImpl(lastPosition.getLeft().getLedgerId(), |
| lastPosition.getLeft().getEntryId() + 1), |
| lastEntryId, cache).get(), |
| Long.valueOf(CompactedTopicImpl.NEWER_THAN_COMPACTED)); |
| |
| // shuffle to make cache work hard |
| Collections.shuffle(positions, r); |
| Collections.shuffle(idsInGaps, r); |
| |
| // Check ids we know are in compacted ledger |
| for (Pair<MessageIdData, Long> p : positions) { |
| PositionImpl pos = new PositionImpl(p.getLeft().getLedgerId(), p.getLeft().getEntryId()); |
| Long got = CompactedTopicImpl.findStartPoint(pos, lastEntryId, cache).get(); |
| Assert.assertEquals(got, p.getRight()); |
| } |
| |
| // Check ids we know are in the gaps of the compacted ledger |
| for (Pair<MessageIdData, Long> gap : idsInGaps) { |
| PositionImpl pos = new PositionImpl(gap.getLeft().getLedgerId(), gap.getLeft().getEntryId()); |
| Assert.assertEquals(CompactedTopicImpl.findStartPoint(pos, lastEntryId, cache).get(), gap.getRight()); |
| } |
| } |
| |
| @Test |
| public void testCleanupOldCompactedTopicLedger() throws Exception { |
| BookKeeper bk = pulsar.getBookKeeperClientFactory().create( |
| this.conf, null, null, Optional.empty(), null); |
| |
| LedgerHandle oldCompactedLedger = bk.createLedger(1, 1, |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD); |
| oldCompactedLedger.close(); |
| LedgerHandle newCompactedLedger = bk.createLedger(1, 1, |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD); |
| newCompactedLedger.close(); |
| |
| // set the compacted topic ledger |
| CompactedTopicImpl compactedTopic = new CompactedTopicImpl(bk); |
| compactedTopic.newCompactedLedger(new PositionImpl(1,2), oldCompactedLedger.getId()).get(); |
| |
| // ensure both ledgers still exist, can be opened |
| bk.openLedger(oldCompactedLedger.getId(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close(); |
| bk.openLedger(newCompactedLedger.getId(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close(); |
| |
| // update the compacted topic ledger |
| PositionImpl newHorizon = new PositionImpl(1,3); |
| compactedTopic.newCompactedLedger(newHorizon, newCompactedLedger.getId()).get(); |
| |
| // Make sure the old compacted ledger still exist after the new compacted ledger created. |
| bk.openLedger(oldCompactedLedger.getId(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close(); |
| |
| Assert.assertTrue(compactedTopic.getCompactedTopicContext().isPresent()); |
| Assert.assertEquals(compactedTopic.getCompactedTopicContext().get().getLedger().getId(), |
| newCompactedLedger.getId()); |
| Assert.assertTrue(compactedTopic.getCompactionHorizon().isPresent()); |
| Assert.assertEquals(compactedTopic.getCompactionHorizon().get(), newHorizon); |
| compactedTopic.deleteCompactedLedger(oldCompactedLedger.getId()).join(); |
| |
| // old ledger should be deleted, new still there |
| try { |
| bk.openLedger(oldCompactedLedger.getId(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close(); |
| Assert.fail("Should have failed to open old ledger"); |
| } catch (BKException.BKNoSuchLedgerExistsException |
| | BKException.BKNoSuchLedgerExistsOnMetadataServerException e) { |
| // correct, expected behaviour |
| } |
| bk.openLedger(newCompactedLedger.getId(), |
| Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, |
| Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close(); |
| } |
| |
| @Test(dataProvider = "batchEnabledProvider") |
| public void testCompactWithEmptyMessage(boolean batchEnabled) throws Exception { |
| final String key = "1"; |
| byte[] msgBytes = "".getBytes(); |
| final String topic = "persistent://my-property/use/my-ns/testCompactWithEmptyMessage-" + UUID.randomUUID(); |
| admin.topics().createPartitionedTopic(topic, 1); |
| final int messages = 10; |
| |
| ProducerBuilder<byte[]> builder = pulsarClient.newProducer().topic(topic); |
| if (!batchEnabled) { |
| builder.enableBatching(false); |
| } else { |
| builder.batchingMaxMessages(messages / 2); |
| } |
| Producer<byte[]> producer = builder.create(); |
| |
| List<CompletableFuture<MessageId>> list = new ArrayList<>(messages); |
| for (int i = 0; i < messages; i++) { |
| list.add(producer.newMessage().keyBytes(key.getBytes(Charset.defaultCharset())).value(msgBytes).sendAsync()); |
| } |
| |
| FutureUtil.waitForAll(list).get(); |
| admin.topics().triggerCompaction(topic); |
| |
| boolean succeed = retryStrategically((test) -> { |
| try { |
| return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status); |
| } catch (PulsarAdminException e) { |
| return false; |
| } |
| }, 10, 200); |
| |
| Assert.assertTrue(succeed); |
| |
| // Send more messages and trigger compaction again. |
| |
| list.clear(); |
| for (int i = 0; i < messages; i++) { |
| list.add(producer.newMessage().key(key).value(msgBytes).sendAsync()); |
| } |
| |
| FutureUtil.waitForAll(list).get(); |
| admin.topics().triggerCompaction(topic); |
| |
| succeed = retryStrategically((test) -> { |
| try { |
| return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status); |
| } catch (PulsarAdminException e) { |
| return false; |
| } |
| }, 10, 200); |
| Assert.assertTrue(succeed); |
| |
| producer.close(); |
| } |
| |
| @Test(timeOut = 30000) |
| public void testReadMessageFromCompactedLedger() throws Exception { |
| final String key = "1"; |
| String msg = "test compaction msg"; |
| final String topic = "persistent://my-property/use/my-ns/testCompactWithEmptyMessage-" + UUID.randomUUID(); |
| admin.topics().createPartitionedTopic(topic, 1); |
| final int numMessages = 10; |
| |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create(); |
| for (int i = 0; i < numMessages; ++i) { |
| producer.newMessage().key(key).value(msg).send(); |
| } |
| |
| admin.topics().triggerCompaction(topic); |
| boolean succeed = retryStrategically((test) -> { |
| try { |
| return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status); |
| } catch (PulsarAdminException e) { |
| return false; |
| } |
| }, 10, 200); |
| |
| Assert.assertTrue(succeed); |
| |
| |
| final String newKey = "2"; |
| String newMsg = "test compaction msg v2"; |
| for (int i = 0; i < numMessages; ++i) { |
| producer.newMessage().key(newKey).value(newMsg).send(); |
| } |
| |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .subscriptionName("test") |
| .readCompacted(true) |
| .startMessageId(MessageId.earliest) |
| .create(); |
| |
| int compactedMsgCount = 0; |
| int nonCompactedMsgCount = 0; |
| while (reader.hasMessageAvailable()) { |
| Message<String> message = reader.readNext(); |
| if (key.equals(message.getKey()) && msg.equals(message.getValue())) { |
| compactedMsgCount++; |
| } else if (newKey.equals(message.getKey()) && newMsg.equals(message.getValue())) { |
| nonCompactedMsgCount++; |
| } |
| } |
| |
| Assert.assertEquals(compactedMsgCount, 1); |
| Assert.assertEquals(nonCompactedMsgCount, numMessages); |
| } |
| |
| @Test |
| public void testLastMessageIdForCompactedLedger() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testLastMessageIdForCompactedLedger-" + UUID.randomUUID(); |
| final String key = "1"; |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create(); |
| final int numMessages = 10; |
| final String msg = "test compaction msg"; |
| for (int i = 0; i < numMessages; ++i) { |
| producer.newMessage().key(key).value(msg).send(); |
| } |
| admin.topics().triggerCompaction(topic); |
| boolean succeed = retryStrategically((test) -> { |
| try { |
| return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status); |
| } catch (PulsarAdminException e) { |
| return false; |
| } |
| }, 10, 200); |
| |
| Assert.assertTrue(succeed); |
| |
| PersistentTopicInternalStats stats0 = admin.topics().getInternalStats(topic); |
| admin.topics().unload(topic); |
| PersistentTopicInternalStats stats1 = admin.topics().getInternalStats(topic); |
| // Make sure the ledger rollover has triggered. |
| Assert.assertTrue(stats0.currentLedgerSize != stats1.currentLedgerSize); |
| |
| Optional<Topic> topicRef = pulsar.getBrokerService().getTopicIfExists(topic).get(); |
| Assert.assertTrue(topicRef.isPresent()); |
| PersistentTopic persistentTopic = (PersistentTopic) topicRef.get(); |
| ManagedLedgerImpl managedLedger = (ManagedLedgerImpl)persistentTopic.getManagedLedger(); |
| managedLedger.maybeUpdateCursorBeforeTrimmingConsumedLedger(); |
| |
| Awaitility.await().untilAsserted(() -> { |
| Assert.assertEquals(managedLedger.getCurrentLedgerEntries(), 0); |
| Assert.assertTrue(managedLedger.getLastConfirmedEntry().getEntryId() != -1); |
| Assert.assertEquals(managedLedger.getLedgersInfoAsList().size(), 1); |
| }); |
| |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .subscriptionName("test") |
| .readCompacted(true) |
| .startMessageId(MessageId.earliest) |
| .create(); |
| |
| Assert.assertTrue(reader.hasMessageAvailable()); |
| Message<String> received = reader.readNext(); |
| Assert.assertEquals(msg, received.getValue()); |
| MessageId messageId = ((ReaderImpl<String>) reader).getConsumer().getLastMessageId(); |
| Assert.assertEquals(messageId, received.getMessageId()); |
| Assert.assertFalse(reader.hasMessageAvailable()); |
| reader.close(); |
| |
| // Unload the topic again to simulate entry ID with -1 after all data has been compacted. |
| admin.topics().unload(topic); |
| PersistentTopicInternalStats stats2 = admin.topics().getInternalStats(topic); |
| Assert.assertTrue(stats2.lastConfirmedEntry.endsWith(":-1")); |
| Assert.assertTrue(stats2.compactedLedger.ledgerId > 0); |
| |
| reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .subscriptionName("test") |
| .readCompacted(true) |
| .startMessageId(MessageId.earliest) |
| .create(); |
| Assert.assertTrue(reader.hasMessageAvailable()); |
| reader.readNext(); |
| Assert.assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| @Test |
| public void testDoNotLossTheLastCompactedLedgerData() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testDoNotLossTheLastCompactedLedgerData-" + |
| UUID.randomUUID(); |
| final int numMessages = 2000; |
| final int keys = 200; |
| final String msg = "Test"; |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .maxPendingMessages(numMessages) |
| .enableBatching(false) |
| .create(); |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i % keys + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, keys); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| admin.topics().unload(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertEquals(stats.ledgers.size(), 1); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| admin.topics().unload(topic); |
| // Send one more key to and then to trigger the compaction |
| producer.newMessage().key(keys + "").value(msg).send(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertEquals(stats.compactedLedger.entries, keys + 1); |
| }); |
| |
| // Make sure the reader can get all data from the compacted ledger and original ledger. |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .create(); |
| int received = 0; |
| while (reader.hasMessageAvailable()) { |
| reader.readNext(); |
| received++; |
| } |
| Assert.assertEquals(received, keys + 1); |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test |
| public void testReadCompactedDataWhenLedgerRolloverKickIn() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testReadCompactedDataWhenLedgerRolloverKickIn-" + |
| UUID.randomUUID(); |
| final int numMessages = 2000; |
| final int keys = 200; |
| final String msg = "Test"; |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .maxPendingMessages(numMessages) |
| .enableBatching(false) |
| .create(); |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i % keys + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, keys); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| // Send more 200 keys |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key((i % keys + keys) + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| |
| // Make sure we have more than 1 original ledgers |
| admin.topics().unload(topic); |
| Awaitility.await().untilAsserted(() -> { |
| Assert.assertEquals(admin.topics().getInternalStats(topic).ledgers.size(), 2); |
| }); |
| |
| // Start a new reader to reading messages |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .receiverQueueSize(10) |
| .create(); |
| |
| // Send more 200 keys |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key((i % keys + keys * 2) + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, keys * 3); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| |
| // The reader should read all 600 keys |
| int received = 0; |
| while (reader.hasMessageAvailable()) { |
| reader.readNext(); |
| received++; |
| } |
| Assert.assertEquals(received, keys * 3); |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test(timeOut = 120000) |
| public void testCompactionWithTopicUnloading() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testCompactionWithTopicUnloading-" + |
| UUID.randomUUID(); |
| final int numMessages = 2000; |
| final int keys = 500; |
| final String msg = "Test"; |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .maxPendingMessages(numMessages) |
| .enableBatching(false) |
| .create(); |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i % keys + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().pollInterval(5, TimeUnit.SECONDS).untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, keys); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| |
| admin.topics().unload(topic); |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key((i % keys + keys) + "").value(msg).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Thread.sleep(100); |
| admin.topics().unload(topic); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().pollInterval(3, TimeUnit.SECONDS).atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, keys * 2); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| }); |
| |
| // Start a new reader to reading messages |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .receiverQueueSize(10) |
| .create(); |
| |
| // The reader should read all 600 keys |
| int received = 0; |
| while (reader.hasMessageAvailable()) { |
| reader.readNext(); |
| received++; |
| } |
| Assert.assertEquals(received, keys * 2); |
| reader.close(); |
| producer.close(); |
| } |
| |
| @Test(timeOut = 1000 * 30) |
| public void testReader() throws Exception { |
| final String ns = "my-property/use/my-ns"; |
| String topic = "persistent://" + ns + "/t1"; |
| |
| @Cleanup |
| Producer<byte[]> producer = pulsarClient.newProducer() |
| .topic(topic) |
| .create(); |
| |
| producer.newMessage().key("k").value(("value").getBytes()).send(); |
| producer.newMessage().key("k").value(null).send(); |
| ((PulsarCompactionServiceFactory)pulsar.getCompactionServiceFactory()).getCompactor().compact(topic).get(); |
| |
| Awaitility.await() |
| .pollInterval(3, TimeUnit.SECONDS) |
| .atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
| admin.topics().unload(topic); |
| Thread.sleep(100); |
| Assert.assertTrue(admin.topics().getInternalStats(topic).lastConfirmedEntry.endsWith("-1")); |
| }); |
| // Make sure the last confirm entry is -1, then get last message id from compact ledger |
| PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic); |
| Assert.assertTrue(internalStats.lastConfirmedEntry.endsWith("-1")); |
| // Because the latest value of the key `k` is null, so there is no data in compact ledger. |
| Assert.assertEquals(internalStats.compactedLedger.size, 0); |
| |
| @Cleanup |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic(topic) |
| .startMessageIdInclusive() |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .create(); |
| Assert.assertFalse(reader.hasMessageAvailable()); |
| } |
| |
| @Test |
| public void testHasMessageAvailableWithNullValueMessage() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testHasMessageAvailable-" + |
| UUID.randomUUID(); |
| final int numMessages = 10; |
| @Cleanup |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .enableBatching(false) |
| .create(); |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i + "").value(String.format("msg [%d]", i)).sendAsync(); |
| } |
| |
| for (int i = numMessages / 2; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i + "").value(null).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, numMessages / 2); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition); |
| }); |
| |
| @Cleanup |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic(topic) |
| .startMessageIdInclusive() |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .create(); |
| for (int i = numMessages / 2; i < numMessages; ++i) { |
| reader.readNext(); |
| } |
| Assert.assertFalse(reader.hasMessageAvailable()); |
| Assert.assertNull(reader.readNext(3, TimeUnit.SECONDS)); |
| } |
| |
| @Test |
| public void testReadCompleteMessagesDuringTopicUnloading() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testReadCompleteMessagesDuringTopicUnloading-" + |
| UUID.randomUUID(); |
| final int numMessages = 1000; |
| @Cleanup |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .enableBatching(false) |
| .create(); |
| // Create a consumer to generate a durable cursor, to avoid deleting the ledger before the reader receives. |
| @Cleanup |
| org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING) |
| .subscriptionName("sub_" + UUID.randomUUID()) |
| .topic(topic) |
| .subscribe(); |
| @Cleanup |
| Reader<String> reader = pulsarClient.newReader(Schema.STRING) |
| .topic(topic) |
| .startMessageIdInclusive() |
| .startMessageId(MessageId.earliest) |
| .readCompacted(true) |
| .create(); |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i + "").value(String.format("msg [%d]", i)).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, numMessages); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition); |
| }); |
| // Unload the topic to make sure the original ledger been deleted. |
| admin.topics().unload(topic); |
| // Produce more messages to the original topic |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i + numMessages + "").value(String.format("msg [%d]", i + numMessages)).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| // For now the topic has 1000 messages in the compacted ledger and 1000 messages in the original topic. |
| // Unloading the topic during reading the data to make sure the reader will not miss any messages. |
| for (int i = 0; i < numMessages / 2; ++i) { |
| Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i)); |
| } |
| admin.topics().unload(topic); |
| for (int i = 0; i < numMessages / 2; ++i) { |
| Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages / 2)); |
| } |
| admin.topics().unload(topic); |
| for (int i = 0; i < numMessages; ++i) { |
| Assert.assertEquals(reader.readNext().getValue(), String.format("msg [%d]", i + numMessages)); |
| } |
| } |
| |
| @Test |
| public void testReadCompactedLatestMessageWithInclusive() throws Exception { |
| String topic = "persistent://my-property/use/my-ns/testLedgerRollover-" + |
| UUID.randomUUID(); |
| final int numMessages = 1; |
| |
| @Cleanup |
| Producer<String> producer = pulsarClient.newProducer(Schema.STRING) |
| .topic(topic) |
| .blockIfQueueFull(true) |
| .enableBatching(false) |
| .create(); |
| |
| CompletableFuture<MessageId> lastMessage = null; |
| for (int i = 0; i < numMessages; ++i) { |
| lastMessage = producer.newMessage().key(i + "").value(String.format("msg [%d]", i)).sendAsync(); |
| } |
| producer.flush(); |
| lastMessage.join(); |
| admin.topics().unload(topic); |
| admin.topics().triggerCompaction(topic); |
| Awaitility.await().untilAsserted(() -> { |
| PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); |
| Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); |
| Assert.assertEquals(stats.compactedLedger.entries, numMessages); |
| Assert.assertEquals(admin.topics().getStats(topic) |
| .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); |
| Assert.assertEquals(stats.lastConfirmedEntry, stats.cursors.get(COMPACTION_SUBSCRIPTION).markDeletePosition); |
| }); |
| |
| Awaitility.await() |
| .pollInterval(3, TimeUnit.SECONDS) |
| .atMost(30, TimeUnit.SECONDS).untilAsserted(() -> { |
| admin.topics().unload(topic); |
| Assert.assertTrue(admin.topics().getInternalStats(topic).lastConfirmedEntry.endsWith("-1")); |
| }); |
| |
| @Cleanup |
| Reader<byte[]> reader = pulsarClient.newReader() |
| .topic(topic) |
| .startMessageIdInclusive() |
| .startMessageId(MessageId.latest) |
| .readCompacted(true) |
| .create(); |
| |
| Assert.assertTrue(reader.hasMessageAvailable()); |
| Assert.assertEquals(reader.readNext().getMessageId(), lastMessage.get()); |
| } |
| } |
