This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-3.0 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b3027fa3e0a410e307016705ea233a59b6f4077d Author: Lari Hotari <[email protected]> AuthorDate: Thu Apr 10 13:22:12 2025 +0300 [fix][test] Fix invalid test CompactionTest.testDeleteCompactedLedgerWithSlowAck (#24166) (cherry picked from commit 1a0f4ba20ef30b07b98aa63dc421c484bc078fb2) --- .../apache/pulsar/compaction/CompactionTest.java | 500 +++++++++++---------- 1 file changed, 255 insertions(+), 245 deletions(-) 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 dc33a02621b..710bde19ff5 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 @@ -18,17 +18,19 @@ */ package org.apache.pulsar.compaction; +import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName; +import static org.apache.pulsar.broker.BrokerTestUtil.spyWithoutRecordingInvocations; import static org.mockito.Mockito.anyLong; -import static org.mockito.Mockito.spy; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertNull; import static org.testng.Assert.assertSame; import static org.testng.Assert.assertThrows; import static org.testng.Assert.assertTrue; - +import static org.testng.Assert.fail; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ThreadFactoryBuilder; import io.netty.buffer.ByteBuf; @@ -45,11 +47,11 @@ import java.util.Random; import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import lombok.Cleanup; import lombok.SneakyThrows; @@ -64,7 +66,6 @@ import org.apache.bookkeeper.mledger.Position; import org.apache.commons.lang3.mutable.MutableLong; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.commons.lang3.tuple.Pair; -import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.namespace.NamespaceService; import org.apache.pulsar.broker.service.Topic; @@ -101,7 +102,6 @@ import org.apache.pulsar.common.protocol.Markers; import org.apache.pulsar.common.util.FutureUtil; import org.awaitility.Awaitility; import org.mockito.Mockito; -import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.DataProvider; @@ -119,10 +119,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { 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"); + admin.clusters().createCluster(configClusterName, + ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); + + admin.tenants().createTenant("my-tenant", + new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Set.of(configClusterName))); + admin.namespaces().createNamespace("my-tenant/my-ns"); compactionScheduler = Executors.newSingleThreadScheduledExecutor( new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build()); @@ -156,7 +158,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testCompaction() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; final int numMessages = 20; final int maxKeys = 10; @@ -185,21 +187,21 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic, false); // Compacted topic ledger should have same number of entry equals to number of unique key. - Assert.assertEquals(expected.size(), internalStats.compactedLedger.entries); - Assert.assertTrue(internalStats.compactedLedger.ledgerId > -1); - Assert.assertFalse(internalStats.compactedLedger.offloaded); + assertEquals(expected.size(), internalStats.compactedLedger.entries); + assertTrue(internalStats.compactedLedger.ledgerId > -1); + assertFalse(internalStats.compactedLedger.offloaded); // consumer with readCompacted enabled only get compacted entries try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { while (true) { Message<byte[]> m = consumer.receive(2, TimeUnit.SECONDS); - Assert.assertEquals(expected.remove(m.getKey()), m.getData()); + assertEquals(expected.remove(m.getKey()), m.getData()); if (expected.isEmpty()) { break; } } - Assert.assertTrue(expected.isEmpty()); + assertTrue(expected.isEmpty()); } // can get full backlog if read compacted disabled @@ -208,24 +210,24 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { while (true) { Message<byte[]> m = consumer.receive(2, TimeUnit.SECONDS); Pair<String, byte[]> expectedMessage = all.remove(0); - Assert.assertEquals(expectedMessage.getLeft(), m.getKey()); - Assert.assertEquals(expectedMessage.getRight(), m.getData()); + assertEquals(expectedMessage.getLeft(), m.getKey()); + assertEquals(expectedMessage.getRight(), m.getData()); if (all.isEmpty()) { break; } } - Assert.assertTrue(all.isEmpty()); + assertTrue(all.isEmpty()); } } @Test public void testCompactionWithReader() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; final int numMessages = 20; final int maxKeys = 10; // Configure retention to ensue data is retained for reader - admin.namespaces().setRetention("my-property/use/my-ns", new RetentionPolicies(-1, -1)); + admin.namespaces().setRetention("my-tenant/my-ns", new RetentionPolicies(-1, -1)); Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -254,12 +256,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .startMessageId(MessageId.earliest).create()) { while (true) { Message<byte[]> m = reader.readNext(2, TimeUnit.SECONDS); - Assert.assertEquals(expected.remove(m.getKey()), new String(m.getData())); + assertEquals(expected.remove(m.getKey()), new String(m.getData())); if (expected.isEmpty()) { break; } } - Assert.assertTrue(expected.isEmpty()); + assertTrue(expected.isEmpty()); } // can get full backlog if read compacted disabled @@ -268,20 +270,20 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { while (true) { Message<byte[]> m = reader.readNext(2, TimeUnit.SECONDS); Pair<String, String> expectedMessage = all.remove(0); - Assert.assertEquals(expectedMessage.getLeft(), m.getKey()); - Assert.assertEquals(expectedMessage.getRight(), new String(m.getData())); + assertEquals(expectedMessage.getLeft(), m.getKey()); + assertEquals(expectedMessage.getRight(), new String(m.getData())); if (all.isEmpty()) { break; } } - Assert.assertTrue(all.isEmpty()); + assertTrue(all.isEmpty()); } } @Test public void testReadCompactedBeforeCompaction() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -297,16 +299,16 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content0".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content0".getBytes()); m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content1".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content1".getBytes()); m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } compact(topic); @@ -314,14 +316,14 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } } @Test public void testReadEntriesAfterCompaction() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -341,18 +343,18 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content3".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content3".getBytes()); } } @Test public void testSeekEarliestAfterCompaction() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -369,8 +371,8 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .readCompacted(true).subscribe()) { consumer.seek(MessageId.earliest); Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -378,22 +380,22 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { consumer.seek(MessageId.earliest); Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content0".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content0".getBytes()); m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content1".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content1".getBytes()); m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } } @Test public void testBrokerRestartAfterCompaction() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -411,15 +413,15 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } stopBroker(); try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { consumer.receive(); - Assert.fail("Shouldn't have been able to receive anything"); + fail("Shouldn't have been able to receive anything"); } catch (PulsarClientException e) { // correct behaviour } @@ -428,14 +430,14 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content2".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content2".getBytes()); } } @Test public void testCompactEmptyTopic() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer() .topic(topic) @@ -451,14 +453,14 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .readCompacted(true).subscribe()) { Message<byte[]> m = consumer.receive(); - Assert.assertEquals(m.getKey(), "key0"); - Assert.assertEquals(m.getData(), "content0".getBytes()); + assertEquals(m.getKey(), "key0"); + assertEquals(m.getData(), "content0".getBytes()); } } @Test public void testFirstMessageRetained() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -486,20 +488,20 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); - Assert.assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); - Assert.assertEquals(message2.getMessageId(), messages.get(2).getMessageId()); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getMessageId(), messages.get(2).getMessageId()); } } @Test public void testBatchMessageIdsDontChange() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -528,13 +530,13 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { } // Ensure all messages are in same batch - Assert.assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getLedgerId(), + assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getLedgerId(), ((BatchMessageIdImpl)messages.get(1).getMessageId()).getLedgerId()); - Assert.assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getLedgerId(), + assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getLedgerId(), ((BatchMessageIdImpl)messages.get(2).getMessageId()).getLedgerId()); - Assert.assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getEntryId(), + assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getEntryId(), ((BatchMessageIdImpl)messages.get(1).getMessageId()).getEntryId()); - Assert.assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getEntryId(), + assertEquals(((BatchMessageIdImpl)messages.get(0).getMessageId()).getEntryId(), ((BatchMessageIdImpl)messages.get(2).getMessageId()).getEntryId()); // compact the topic @@ -544,25 +546,25 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); if (getCompactor() instanceof StrategicTwoPhaseCompactor) { - Assert.assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); - Assert.assertEquals(message2.getMessageId(), messages.get(1).getMessageId()); + assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); + assertEquals(message2.getMessageId(), messages.get(1).getMessageId()); } else { - Assert.assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); - Assert.assertEquals(message2.getMessageId(), messages.get(2).getMessageId()); + assertEquals(message1.getMessageId(), messages.get(0).getMessageId()); + assertEquals(message2.getMessageId(), messages.get(2).getMessageId()); } } } @Test public void testBatchMessageWithNullValue() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") .receiverQueueSize(1).readCompacted(true).subscribe().close(); @@ -616,7 +618,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testWholeBatchCompactedOut() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -645,8 +647,8 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message = consumer.receive(); - Assert.assertEquals(message.getKey(), "key1"); - Assert.assertEquals(new String(message.getData()), "my-message-4"); + assertEquals(message.getKey(), "key1"); + assertEquals(new String(message.getData()), "my-message-4"); } } @@ -661,7 +663,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { restartBroker(); FieldUtils.writeDeclaredField(compactor, "topicCompactionRetainNullKey", retainNullKey, true); - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -691,12 +693,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .subscriptionName("sub1").readCompacted(true).subscribe()){ if (getCompactor() instanceof StrategicTwoPhaseCompactor) { Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key1"); - Assert.assertEquals(new String(message3.getData()), "my-message-4"); + assertEquals(message3.getKey(), "key1"); + assertEquals(new String(message3.getData()), "my-message-4"); Message<byte[]> message4 = consumer.receive(); - Assert.assertEquals(message4.getKey(), "key2"); - Assert.assertEquals(new String(message4.getData()), "my-message-6"); + assertEquals(message4.getKey(), "key2"); + assertEquals(new String(message4.getData()), "my-message-6"); Message<byte[]> m = consumer.receive(2, TimeUnit.SECONDS); assertNull(m); @@ -719,7 +721,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { } else { expectList = List.of(Pair.of("key1", "my-message-4"), Pair.of("key2", "my-message-6")); } - Assert.assertEquals(result, expectList); + assertEquals(result, expectList); } } } @@ -727,7 +729,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testEmptyPayloadDeletes() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -798,18 +800,18 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key0"); - Assert.assertEquals(new String(message1.getData()), "my-message-0"); + assertEquals(message1.getKey(), "key0"); + assertEquals(new String(message1.getData()), "my-message-0"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key4"); - Assert.assertEquals(new String(message2.getData()), "my-message-4"); + assertEquals(message2.getKey(), "key4"); + assertEquals(new String(message2.getData()), "my-message-4"); } } @Test public void testEmptyPayloadDeletesWhenCompressed() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -877,12 +879,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key0"); - Assert.assertEquals(new String(message1.getData()), "my-message-0"); + assertEquals(message1.getKey(), "key0"); + assertEquals(new String(message1.getData()), "my-message-0"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key4"); - Assert.assertEquals(new String(message2.getData()), "my-message-4"); + assertEquals(message2.getKey(), "key4"); + assertEquals(new String(message2.getData()), "my-message-4"); } } @@ -890,13 +892,13 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testCompactorReadsCompacted() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // capture opened ledgers Set<Long> ledgersOpened = Sets.newConcurrentHashSet(); when(pulsarTestContext.getBookKeeperClient().newOpenLedgerOp()).thenAnswer( (invocation) -> { - OpenBuilder builder = (OpenBuilder)spy(invocation.callRealMethod()); + OpenBuilder builder = (OpenBuilder)spyWithoutRecordingInvocations(invocation.callRealMethod()); when(builder.withLedgerId(anyLong())).thenAnswer( (invocation2) -> { ledgersOpened.add((Long)invocation2.getArguments()[0]); @@ -931,15 +933,15 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { String managedLedgerName = ((PersistentTopic)pulsar.getBrokerService().getTopicReference(topic).get()) .getManagedLedger().getName(); ManagedLedgerInfo info = pulsar.getManagedLedgerFactory().getManagedLedgerInfo(managedLedgerName); - Assert.assertEquals(info.ledgers.size(), 2); - Assert.assertTrue(ledgersOpened.isEmpty()); // no ledgers should have been opened + assertEquals(info.ledgers.size(), 2); + assertTrue(ledgersOpened.isEmpty()); // no ledgers should have been opened // compact the topic compact(topic); // should have opened all except last to read - Assert.assertTrue(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); + assertTrue(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); ledgersOpened.clear(); // force broker to close resources for topic @@ -954,12 +956,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { } info = pulsar.getManagedLedgerFactory().getManagedLedgerInfo(managedLedgerName); - Assert.assertEquals(info.ledgers.size(), 3); + assertEquals(info.ledgers.size(), 3); // should only have opened the penultimate ledger to get stat - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(2).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(2).ledgerId)); ledgersOpened.clear(); // compact the topic again @@ -967,30 +969,30 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { // shouldn't have opened first ledger (already compacted), penultimate would have some uncompacted data. // last ledger already open for writing - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); - Assert.assertTrue(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); - Assert.assertFalse(ledgersOpened.contains(info.ledgers.get(2).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(0).ledgerId)); + assertTrue(ledgersOpened.contains(info.ledgers.get(1).ledgerId)); + assertFalse(ledgersOpened.contains(info.ledgers.get(2).ledgerId)); // all three messages should be there when we read compacted try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key0"); - Assert.assertEquals(new String(message1.getData()), "my-message-0"); + assertEquals(message1.getKey(), "key0"); + assertEquals(new String(message1.getData()), "my-message-0"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key1"); - Assert.assertEquals(new String(message2.getData()), "my-message-1"); + assertEquals(message2.getKey(), "key1"); + assertEquals(new String(message2.getData()), "my-message-1"); Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key2"); - Assert.assertEquals(new String(message3.getData()), "my-message-2"); + assertEquals(message3.getKey(), "key2"); + assertEquals(new String(message3.getData()), "my-message-2"); } } @Test public void testCompactCompressedNoBatch() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1018,18 +1020,18 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); } } @Test public void testCompactCompressedBatching() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1061,12 +1063,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { try (Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); } } @@ -1081,10 +1083,10 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); return keyInfo; } catch (IOException e) { - Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); + fail("Failed to read certificate from " + CERT_FILE_PATH); } } else { - Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); + fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); } return null; } @@ -1097,10 +1099,10 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { keyInfo.setKey(Files.readAllBytes(Paths.get(CERT_FILE_PATH))); return keyInfo; } catch (IOException e) { - Assert.fail("Failed to read certificate from " + CERT_FILE_PATH); + fail("Failed to read certificate from " + CERT_FILE_PATH); } } else { - Assert.fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); + fail("Certificate file " + CERT_FILE_PATH + " is not present or not readable."); } return null; } @@ -1108,7 +1110,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testCompactEncryptedNoBatch() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1139,18 +1141,18 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .subscriptionName("sub1").cryptoKeyReader(new EncKeyReader()) .readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); } } @Test public void testCompactEncryptedBatching() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1183,30 +1185,30 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .subscriptionName("sub1").cryptoKeyReader(new EncKeyReader()) .readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); if (getCompactor() instanceof StrategicTwoPhaseCompactor) { Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key2"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key2"); + assertEquals(new String(message3.getData()), "my-message-3"); } else { // with encryption, all messages are passed through compaction as it doesn't // have the keys to decrypt the batch payload Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-2"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-2"); Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key2"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key2"); + assertEquals(new String(message3.getData()), "my-message-3"); } } } @Test public void testCompactEncryptedAndCompressedNoBatch() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1238,18 +1240,18 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .subscriptionName("sub1").cryptoKeyReader(new EncKeyReader()) .readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-3"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-3"); } } @Test public void testCompactEncryptedAndCompressedBatching() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1285,29 +1287,29 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .subscriptionName("sub1").cryptoKeyReader(new EncKeyReader()) .readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key1"); - Assert.assertEquals(new String(message1.getData()), "my-message-1"); + assertEquals(message1.getKey(), "key1"); + assertEquals(new String(message1.getData()), "my-message-1"); if (getCompactor() instanceof StrategicTwoPhaseCompactor) { Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key2"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key2"); + assertEquals(new String(message3.getData()), "my-message-3"); } else { Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-2"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-2"); Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key2"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key2"); + assertEquals(new String(message3.getData()), "my-message-3"); } } } @Test public void testEmptyPayloadDeletesWhenEncrypted() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; // subscribe before sending anything, so that we get all messages pulsarClient.newConsumer().topic(topic).subscriptionName("sub1") @@ -1364,34 +1366,34 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .cryptoKeyReader(new EncKeyReader()) .subscriptionName("sub1").readCompacted(true).subscribe()){ Message<byte[]> message1 = consumer.receive(); - Assert.assertEquals(message1.getKey(), "key0"); - Assert.assertEquals(new String(message1.getData()), "my-message-0"); + assertEquals(message1.getKey(), "key0"); + assertEquals(new String(message1.getData()), "my-message-0"); if (getCompactor() instanceof StrategicTwoPhaseCompactor) { Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key3"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key3"); + assertEquals(new String(message3.getData()), "my-message-3"); Message<byte[]> message5 = consumer.receive(); - Assert.assertEquals(message5.getKey(), "key4"); - Assert.assertEquals(new String(message5.getData()), "my-message-4"); + assertEquals(message5.getKey(), "key4"); + assertEquals(new String(message5.getData()), "my-message-4"); } else { // see all messages from batch Message<byte[]> message2 = consumer.receive(); - Assert.assertEquals(message2.getKey(), "key2"); - Assert.assertEquals(new String(message2.getData()), "my-message-2"); + assertEquals(message2.getKey(), "key2"); + assertEquals(new String(message2.getData()), "my-message-2"); Message<byte[]> message3 = consumer.receive(); - Assert.assertEquals(message3.getKey(), "key3"); - Assert.assertEquals(new String(message3.getData()), "my-message-3"); + assertEquals(message3.getKey(), "key3"); + assertEquals(new String(message3.getData()), "my-message-3"); Message<byte[]> message4 = consumer.receive(); - Assert.assertEquals(message4.getKey(), "key2"); + assertEquals(message4.getKey(), "key2"); assertNull(message4.getData()); Message<byte[]> message5 = consumer.receive(); - Assert.assertEquals(message5.getKey(), "key4"); - Assert.assertEquals(new String(message5.getData()), "my-message-4"); + assertEquals(message5.getKey(), "key4"); + assertEquals(new String(message5.getData()), "my-message-4"); } } } @@ -1403,7 +1405,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000, dataProvider = "lastDeletedBatching") public void testCompactionWithLastDeletedKey(boolean batching) throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(batching) .messageRoutingMode(MessageRoutingMode.SinglePartition).create(); @@ -1429,7 +1431,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000, dataProvider = "lastDeletedBatching") public void testEmptyCompactionLedger(boolean batching) throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic1"; + String topic = "persistent://my-tenant/my-ns/my-topic1"; Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(batching) .messageRoutingMode(MessageRoutingMode.SinglePartition).create(); @@ -1453,7 +1455,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000, dataProvider = "lastDeletedBatching") public void testAllEmptyCompactionLedger(boolean batchEnabled) throws Exception { - final String topic = "persistent://my-property/use/my-ns/testAllEmptyCompactionLedger" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testAllEmptyCompactionLedger" + UUID.randomUUID().toString(); final int messages = 10; @@ -1487,7 +1489,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000) public void testBatchAndNonBatchWithoutEmptyPayload() throws PulsarClientException, ExecutionException, InterruptedException { - final String topic = "persistent://my-property/use/my-ns/testBatchAndNonBatchWithoutEmptyPayload" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testBatchAndNonBatchWithoutEmptyPayload" + UUID.randomUUID().toString(); // 1.create producer and publish message to the topic. Producer<byte[]> producer = pulsarClient.newProducer() @@ -1538,7 +1540,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { } @Test(timeOut = 20000) public void testBatchAndNonBatchWithEmptyPayload() throws PulsarClientException, ExecutionException, InterruptedException { - final String topic = "persistent://my-property/use/my-ns/testBatchAndNonBatchWithEmptyPayload" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testBatchAndNonBatchWithEmptyPayload" + UUID.randomUUID().toString(); // 1.create producer and publish message to the topic. Producer<byte[]> producer = pulsarClient.newProducer() @@ -1594,7 +1596,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000) public void testBatchAndNonBatchEndOfEmptyPayload() throws PulsarClientException, ExecutionException, InterruptedException { - final String topic = "persistent://my-property/use/my-ns/testBatchAndNonBatchWithEmptyPayload" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testBatchAndNonBatchWithEmptyPayload" + UUID.randomUUID().toString(); // 1.create producer and publish message to the topic. Producer<byte[]> producer = pulsarClient.newProducer() @@ -1643,7 +1645,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 20000, dataProvider = "lastDeletedBatching") public void testCompactMultipleTimesWithoutEmptyMessage(boolean batchEnabled) throws PulsarClientException, ExecutionException, InterruptedException { - final String topic = "persistent://my-property/use/my-ns/testCompactMultipleTimesWithoutEmptyMessage" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testCompactMultipleTimesWithoutEmptyMessage" + UUID.randomUUID().toString(); final int messages = 10; final String key = "1"; @@ -1691,7 +1693,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 2000000, dataProvider = "lastDeletedBatching") public void testReadUnCompacted(boolean batchEnabled) throws PulsarClientException, ExecutionException, InterruptedException { - final String topic = "persistent://my-property/use/my-ns/testReadUnCompacted" + UUID.randomUUID().toString(); + final String topic = "persistent://my-tenant/my-ns/testReadUnCompacted" + UUID.randomUUID().toString(); final int messages = 10; final String key = "1"; @@ -1797,9 +1799,9 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test(timeOut = 60000) public void testCompactionWithMarker() throws Exception { - String namespace = "my-property/use/my-ns"; + String namespace = "my-tenant/my-ns"; final TopicName dest = TopicName.get( - BrokerTestUtil.newUniqueName("persistent://" + namespace + "/testWriteMarker")); + newUniqueName("persistent://" + namespace + "/testWriteMarker")); admin.topics().createNonPartitionedTopic(dest.toString()); @Cleanup Consumer<byte[]> consumer = pulsarClient.newConsumer() @@ -1817,7 +1819,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .create(); producer.send("msg-1".getBytes(StandardCharsets.UTF_8)); Optional<Topic> topic = pulsar.getBrokerService().getTopic(dest.toString(), true).join(); - Assert.assertTrue(topic.isPresent()); + assertTrue(topic.isPresent()); PersistentTopic persistentTopic = (PersistentTopic) topic.get(); Random random = new Random(); for (int i = 0; i < 100; i++) { @@ -1850,13 +1852,13 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .pollInterval(1, TimeUnit.SECONDS) .untilAsserted(() -> { long ledgerId = admin.topics().getInternalStats(dest.toString()).compactedLedger.ledgerId; - Assert.assertNotEquals(ledgerId, -1L); + assertNotEquals(ledgerId, -1L); }); } @Test(timeOut = 100000) public void testReceiverQueueSize() throws Exception { - final String topicName = "persistent://my-property/use/my-ns/testReceiverQueueSize" + UUID.randomUUID(); + final String topicName = newUniqueName("persistent://my-tenant/my-ns/testReceiverQueueSize"); final String subName = "my-sub"; final int receiveQueueSize = 1; @Cleanup @@ -1883,7 +1885,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { //Give some time to consume Awaitility.await() - .untilAsserted(() -> Assert.assertEquals(consumer.getStats().getMsgNumInReceiverQueue().intValue(), + .untilAsserted(() -> assertEquals(consumer.getStats().getMsgNumInReceiverQueue().intValue(), receiveQueueSize)); consumer.close(); producer.close(); @@ -1892,7 +1894,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testDispatcherMaxReadSizeBytes() throws Exception { final String topicName = - "persistent://my-property/use/my-ns/testDispatcherMaxReadSizeBytes" + UUID.randomUUID(); + newUniqueName("persistent://my-tenant/my-ns/testDispatcherMaxReadSizeBytes"); final String subName = "my-sub"; final int receiveQueueSize = 1; @Cleanup @@ -1943,7 +1945,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testCompactionDuplicate() throws Exception { - String topic = "persistent://my-property/use/my-ns/testCompactionDuplicate"; + String topic = "persistent://my-tenant/my-ns/testCompactionDuplicate"; final int numMessages = 1000; final int maxKeys = 800; @@ -1989,9 +1991,9 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { Awaitility.await().untilAsserted(() -> { PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic, false); // Compacted topic ledger should have same number of entry equals to number of unique key. - Assert.assertEquals(internalStats.compactedLedger.entries, expected.size()); - Assert.assertTrue(internalStats.compactedLedger.ledgerId > -1); - Assert.assertFalse(internalStats.compactedLedger.offloaded); + assertEquals(internalStats.compactedLedger.entries, expected.size()); + assertTrue(internalStats.compactedLedger.ledgerId > -1); + assertFalse(internalStats.compactedLedger.offloaded); }); // consumer with readCompacted enabled only get compacted entries @@ -1999,7 +2001,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { .readCompacted(true).subscribe()) { while (true) { Message<byte[]> m = consumer.receive(2, TimeUnit.SECONDS); - Assert.assertEquals(expected.remove(m.getKey()), m.getData()); + assertEquals(expected.remove(m.getKey()), m.getData()); if (expected.isEmpty()) { break; } @@ -2009,7 +2011,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testDeleteCompactedLedger() throws Exception { - String topicName = "persistent://my-property/use/my-ns/testDeleteCompactedLedger"; + String topicName = "persistent://my-tenant/my-ns/testDeleteCompactedLedger"; final String subName = "my-sub"; @Cleanup @@ -2028,9 +2030,9 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { MutableLong compactedLedgerId = new MutableLong(-1); Awaitility.await().untilAsserted(() -> { PersistentTopicInternalStats stats = admin.topics().getInternalStats(topicName); - Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1L); + assertNotEquals(stats.compactedLedger.ledgerId, -1L); compactedLedgerId.setValue(stats.compactedLedger.ledgerId); - Assert.assertEquals(stats.compactedLedger.entries, 2L); + assertEquals(stats.compactedLedger.entries, 2L); }); // delete compacted ledger @@ -2038,8 +2040,8 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { Awaitility.await().untilAsserted(() -> { PersistentTopicInternalStats stats = admin.topics().getInternalStats(topicName); - Assert.assertEquals(stats.compactedLedger.ledgerId, -1L); - Assert.assertEquals(stats.compactedLedger.entries, -1L); + assertEquals(stats.compactedLedger.ledgerId, -1L); + assertEquals(stats.compactedLedger.entries, -1L); assertThrows(BKException.BKNoSuchLedgerExistsException.class, () -> pulsarTestContext.getBookKeeperClient() .openLedger(compactedLedgerId.getValue(), BookKeeper.DigestType.CRC32C, new byte[]{})); }); @@ -2049,9 +2051,9 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { MutableLong compactedLedgerId2 = new MutableLong(-1); Awaitility.await().untilAsserted(() -> { PersistentTopicInternalStats stats = admin.topics().getInternalStats(topicName); - Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1L); + assertNotEquals(stats.compactedLedger.ledgerId, -1L); compactedLedgerId2.setValue(stats.compactedLedger.ledgerId); - Assert.assertEquals(stats.compactedLedger.entries, 2L); + assertEquals(stats.compactedLedger.entries, 2L); }); producer.close(); @@ -2062,77 +2064,85 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { compactedLedgerId2.getValue(), BookKeeper.DigestType.CRC32, new byte[]{}))); } - @Test + @Test(timeOut = 10000) public void testDeleteCompactedLedgerWithSlowAck() throws Exception { - // Disable topic level policies, since block ack thread may also block thread of delete topic policies. - conf.setTopicLevelPoliciesEnabled(false); - restartBroker(); + String topicName = newUniqueName("persistent://my-tenant/my-ns/testDeleteCompactedLedgerWithSlowAck"); + admin.topics().createNonPartitionedTopic(topicName); + // minimum compaction threshold + admin.topicPolicies().setCompactionThreshold(topicName, 1); + // infinite retention + admin.topicPolicies().setRetention(topicName, new RetentionPolicies(-1, -1)); - String topicName = "persistent://my-property/use/my-ns/testDeleteCompactedLedgerWithSlowAck"; @Cleanup Producer<String> producer = pulsarClient.newProducer(Schema.STRING) .enableBatching(false).topic(topicName).create(); - pulsarClient.newConsumer().topic(topicName).subscriptionType(SubscriptionType.Exclusive) - .subscriptionName(Compactor.COMPACTION_SUBSCRIPTION) - .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).readCompacted(true).subscribe() - .close(); + // send a single message + producer.newMessage().key(String.valueOf(0)).value("0").send(); + // trigger compaction once to create __compaction subscription + triggerCompactionAndWait(topicName); - for (int i = 0; i < 10; i++) { + int numberOfMessages = 10; + for (int i = 0; i < numberOfMessages; i++) { producer.newMessage().key(String.valueOf(i % 2)).value(String.valueOf(i)).sendAsync(); } producer.flush(); + // replace the PersistentSubscription with a spy PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get(); - PersistentSubscription subscription = spy(topic.getSubscription(Compactor.COMPACTION_SUBSCRIPTION)); + PersistentSubscription subscription = + spyWithoutRecordingInvocations(topic.getSubscription(Compactor.COMPACTION_SUBSCRIPTION)); topic.getSubscriptions().put(Compactor.COMPACTION_SUBSCRIPTION, subscription); + // delay the ack of compaction + CountDownLatch compactionAckedLatch = new CountDownLatch(1); AtomicLong compactedLedgerId = new AtomicLong(-1); - AtomicBoolean pauseAck = new AtomicBoolean(); Mockito.doAnswer(invocationOnMock -> { - Map<String, Long> properties = (Map<String, Long>) invocationOnMock.getArguments()[2]; - log.info("acknowledgeMessage properties: {}", properties); + List<Position> positions = invocationOnMock.getArgument(0); + Map<String, Long> properties = invocationOnMock.getArgument(2); + log.info("acknowledgeMessage positions: {} properties: {}", positions, properties); compactedLedgerId.set(properties.get(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY)); - pauseAck.set(true); - while (pauseAck.get()) { - Thread.sleep(200); + try { + return invocationOnMock.callRealMethod(); + } finally { + log.info("acknowledgeMessage completed {}", positions); + compactionAckedLatch.countDown(); + // add delay here to introduce possible races with deletion + Thread.sleep(500); } - return invocationOnMock.callRealMethod(); }).when(subscription).acknowledgeMessage(Mockito.any(), Mockito.eq( CommandAck.AckType.Cumulative), Mockito.any()); + // trigger compaction admin.topics().triggerCompaction(topicName); - while (!pauseAck.get()) { - Thread.sleep(100); - } + // wait for compaction to acknowledge + compactionAckedLatch.await(9, TimeUnit.SECONDS); - CompletableFuture<Long> currentCompaction = - (CompletableFuture<Long>) FieldUtils.readDeclaredField(topic, "currentCompaction", true); - CompletableFuture<Long> spyCurrentCompaction = spy(currentCompaction); - FieldUtils.writeDeclaredField(topic, "currentCompaction", spyCurrentCompaction, true); - currentCompaction.whenComplete((obj, throwable) -> { - if (throwable != null) { - spyCurrentCompaction.completeExceptionally(throwable); - } else { - spyCurrentCompaction.complete(obj); - } - }); - Mockito.doAnswer(invocationOnMock -> { - pauseAck.set(false); - return invocationOnMock.callRealMethod(); - }).when(spyCurrentCompaction).handle(Mockito.any()); + // close the producer + producer.close(); + // delete compacted ledger admin.topics().delete(topicName, true); + // ensure that the compacted ledger is deleted Awaitility.await().untilAsserted(() -> assertThrows(BKException.BKNoSuchLedgerExistsException.class, () -> pulsarTestContext.getBookKeeperClient().openLedger( compactedLedgerId.get(), BookKeeper.DigestType.CRC32, new byte[]{}))); } + private void triggerCompactionAndWait(String topicName) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + persistentTopic.triggerCompaction(); + CompletableFuture<Long> currentCompaction = + (CompletableFuture<Long>) FieldUtils.readDeclaredField(persistentTopic, "currentCompaction", true); + currentCompaction.get(10, TimeUnit.SECONDS); + } + @Test public void testCompactionWithTTL() throws Exception { - String topicName = "persistent://my-property/use/my-ns/testCompactionWithTTL"; + String topicName = "persistent://my-tenant/my-ns/testCompactionWithTTL"; String subName = "sub"; pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName).readCompacted(true) .subscribe().close(); @@ -2193,12 +2203,12 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { result.add(receive.getValue()); } - Assert.assertEquals(result, List.of("V3", "V4", "V5")); + assertEquals(result, List.of("V3", "V4", "V5")); } @Test public void testAcknowledgeWithReconnection() throws Exception { - final String topicName = "persistent://my-property/use/my-ns/testAcknowledge" + UUID.randomUUID(); + final String topicName = newUniqueName("persistent://my-tenant/my-ns/testAcknowledge"); final String subName = "my-sub"; @Cleanup PulsarClient client = newPulsarClient(lookupUrl.toString(), 100); @@ -2253,7 +2263,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { // Wait for consumer to reconnect and clear incomingMessages consumer.pause(); Awaitility.await().untilAsserted(() -> { - Assert.assertEquals(consumer.numMessagesInQueue(), 0); + assertEquals(consumer.numMessagesInQueue(), 0); }); consumer.resume(); @@ -2270,22 +2280,22 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { assertEquals(admin.topics().getStats(topicName, true).getSubscriptions().get(subName).getMsgBacklog(), 0)); - Assert.assertEquals(results, expected); + assertEquals(results, expected); Message<String> message = consumer.receive(3, TimeUnit.SECONDS); - Assert.assertNull(message); + assertNull(message); // Make consumer reconnect to broker admin.topics().unload(topicName); producer.newMessage().key("K").value("V").send(); Message<String> message2 = consumer.receive(3, TimeUnit.SECONDS); - Assert.assertEquals(message2.getValue(), "V"); + assertEquals(message2.getValue(), "V"); consumer.acknowledge(message2); Awaitility.await().untilAsserted(() -> { PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topicName); - Assert.assertEquals(internalStats.lastConfirmedEntry, + assertEquals(internalStats.lastConfirmedEntry, internalStats.cursors.get(subName).markDeletePosition); }); @@ -2295,7 +2305,7 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { @Test public void testEarliestSubsAfterRollover() throws Exception { - final String topicName = "persistent://my-property/use/my-ns/testEarliestSubsAfterRollover" + UUID.randomUUID(); + final String topicName = newUniqueName("persistent://my-tenant/my-ns/testEarliestSubsAfterRollover"); final String subName = "my-sub"; @Cleanup PulsarClient client = newPulsarClient(lookupUrl.toString(), 100); @@ -2347,6 +2357,6 @@ public class CompactionTest extends MockedPulsarServiceBaseTest { consumer.acknowledge(message); } - Assert.assertEquals(results, expected); + assertEquals(results, expected); } }
