Skip to content

Commit 765c46e

Browse files
merlimatlhotari
merlimat
and
lhotari
authored
[cleanup] Convert 10 test classes to SharedPulsarBaseTest (#25338)
Co-authored-by: Lari Hotari <lhotari@users.noreply.github.com>
1 parent 86e9e64 commit 765c46e

14 files changed

Lines changed: 376 additions & 480 deletions

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ConsumedLedgersTrimTest.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ public class ConsumedLedgersTrimTest extends SharedPulsarBaseTest {
5050

5151
@Test
5252
public void testConsumedLedgersTrim() throws Exception {
53+
// Disable dedup so the pulsar.dedup cursor doesn't block ledger trimming
54+
admin.namespaces().setDeduplicationStatus(getNamespace(), false);
5355
// Set infinite retention at namespace level so ledgers are preserved until explicitly trimmed
5456
admin.namespaces().setRetention(getNamespace(), new RetentionPolicies(-1, -1));
5557

@@ -106,6 +108,8 @@ public void testConsumedLedgersTrim() throws Exception {
106108

107109
@Test
108110
public void testConsumedLedgersTrimNoSubscriptions() throws Exception {
111+
// Disable dedup so the pulsar.dedup cursor doesn't block ledger trimming
112+
admin.namespaces().setDeduplicationStatus(getNamespace(), false);
109113
// Set infinite retention at namespace level so ledgers are preserved until explicitly trimmed
110114
admin.namespaces().setRetention(getNamespace(), new RetentionPolicies(-1, -1));
111115

@@ -178,6 +182,8 @@ public void testConsumedLedgersTrimNoSubscriptions() throws Exception {
178182

179183
@Test
180184
public void testAdminTrimLedgers() throws Exception {
185+
// Disable dedup so the pulsar.dedup cursor doesn't block ledger trimming
186+
admin.namespaces().setDeduplicationStatus(getNamespace(), false);
181187
final String subscriptionName = "my-sub";
182188
final int maxEntriesPerLedger = 2;
183189
final int partitionedNum = 3;
@@ -230,7 +236,6 @@ public void testAdminTrimLedgers() throws Exception {
230236
"Expected at most 2 ledgers after trim, but found "
231237
+ managedLedger.getLedgersInfoAsList().size());
232238
});
233-
234239
}
235240

236241
@Test

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CurrentLedgerRolloverIfFullTest.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@ public class CurrentLedgerRolloverIfFullTest extends SharedPulsarBaseTest {
3737

3838
@Test
3939
public void testCurrentLedgerRolloverIfFull() throws Exception {
40+
// Disable dedup so the pulsar.dedup cursor doesn't block ledger trimming
41+
admin.namespaces().setDeduplicationStatus(getNamespace(), false);
4042
final String topicName = newTopicName();
4143

4244
@Cleanup

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SharedPulsarCluster.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@ private void start() throws Exception {
146146
config.setForceDeleteNamespaceAllowed(true);
147147
config.setForceDeleteTenantAllowed(true);
148148
config.setBrokerDeleteInactiveTopicsEnabled(false);
149+
config.setBrokerDeduplicationEnabled(true);
149150

150151
// Reduce thread pool sizes for faster startup (fewer threads to create)
151152
config.setNumIOThreads(2);

‎pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientDeduplicationTest.java‎

Lines changed: 27 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -34,18 +34,17 @@
3434
import java.util.stream.Collectors;
3535
import lombok.Cleanup;
3636
import lombok.extern.slf4j.Slf4j;
37+
import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
3738
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
3839
import org.apache.pulsar.client.impl.MessageIdImpl;
3940
import org.apache.pulsar.common.util.FutureUtil;
4041
import org.awaitility.Awaitility;
41-
import org.testng.annotations.AfterClass;
42-
import org.testng.annotations.BeforeClass;
4342
import org.testng.annotations.DataProvider;
4443
import org.testng.annotations.Test;
4544

4645
@Slf4j
4746
@Test(groups = "broker-api")
48-
public class ClientDeduplicationTest extends ProducerConsumerBase {
47+
public class ClientDeduplicationTest extends SharedPulsarBaseTest {
4948

5049
@DataProvider
5150
public static Object[][] batchingTypes() {
@@ -55,22 +54,9 @@ public static Object[][] batchingTypes() {
5554
};
5655
}
5756

58-
@BeforeClass
59-
@Override
60-
protected void setup() throws Exception {
61-
super.internalSetup();
62-
super.producerBaseSetup();
63-
}
64-
65-
@AfterClass(alwaysRun = true)
66-
@Override
67-
protected void cleanup() throws Exception {
68-
super.internalCleanup();
69-
}
70-
7157
@Test(priority = -1)
7258
public void testNamespaceDeduplicationApi() throws Exception {
73-
final String namespace = "my-property/my-ns";
59+
final String namespace = getNamespace();
7460
assertNull(admin.namespaces().getDeduplicationStatus(namespace));
7561
admin.namespaces().setDeduplicationStatus(namespace, true);
7662
Awaitility.await().untilAsserted(() -> assertTrue(admin.namespaces().getDeduplicationStatus(namespace)));
@@ -82,8 +68,8 @@ public void testNamespaceDeduplicationApi() throws Exception {
8268

8369
@Test
8470
public void testProducerSequenceAfterReconnect() throws Exception {
85-
final String topic = "persistent://my-property/my-ns/testProducerSequenceAfterReconnect";
86-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
71+
final String topic = newTopicName();
72+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
8773

8874
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
8975
.producerName("my-producer-name");
@@ -113,8 +99,8 @@ public void testProducerSequenceAfterReconnect() throws Exception {
11399

114100
@Test
115101
public void testProducerSequenceAfterRestart() throws Exception {
116-
String topic = "persistent://my-property/my-ns/testProducerSequenceAfterRestart";
117-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
102+
String topic = newTopicName();
103+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
118104

119105
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
120106
.producerName("my-producer-name");
@@ -130,8 +116,8 @@ public void testProducerSequenceAfterRestart() throws Exception {
130116

131117
producer.close();
132118

133-
// Kill and restart broker
134-
restartBroker();
119+
// Unload topic to force reload of dedup state
120+
admin.topics().unload(topic);
135121

136122
producer = producerBuilder.create();
137123
assertEquals(producer.getLastSequenceId(), 9L);
@@ -147,8 +133,8 @@ public void testProducerSequenceAfterRestart() throws Exception {
147133

148134
@Test(timeOut = 30000)
149135
public void testProducerDeduplication() throws Exception {
150-
String topic = "persistent://my-property/my-ns/testProducerDeduplication";
151-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
136+
String topic = newTopicName();
137+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
152138

153139
// Set infinite timeout
154140
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
@@ -180,8 +166,8 @@ public void testProducerDeduplication() throws Exception {
180166
Message<byte[]> msg = consumer.receive(1, TimeUnit.SECONDS);
181167
assertNull(msg);
182168

183-
// Kill and restart broker
184-
restartBroker();
169+
// Unload topic to force reload of dedup state
170+
admin.topics().unload(topic);
185171

186172
producer = producerBuilder.create();
187173
assertEquals(producer.getLastSequenceId(), 2L);
@@ -198,9 +184,8 @@ public void testProducerDeduplication() throws Exception {
198184

199185
@Test(timeOut = 30000, dataProvider = "batchingTypes")
200186
public void testProducerDeduplicationWithDiscontinuousSequenceId(BatcherBuilder batcherBuilder) throws Exception {
201-
String topic = "persistent://my-property/my-ns/testProducerDeduplicationWithDiscontinuousSequenceId-"
202-
+ System.currentTimeMillis();
203-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
187+
String topic = newTopicName();
188+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
204189

205190
// Set infinite timeout
206191
ProducerBuilder<byte[]> producerBuilder =
@@ -244,8 +229,8 @@ public void testProducerDeduplicationWithDiscontinuousSequenceId(BatcherBuilder
244229
assertNull(msg);
245230

246231
producer.close();
247-
// Kill and restart broker
248-
restartBroker();
232+
// Unload topic to force reload of dedup state
233+
admin.topics().unload(topic);
249234

250235
producer = producerBuilder.create();
251236
assertEquals(producer.getLastSequenceId(), 6L);
@@ -263,8 +248,8 @@ public void testProducerDeduplicationWithDiscontinuousSequenceId(BatcherBuilder
263248

264249
@Test(timeOut = 30000)
265250
public void testProducerDeduplicationNonBatchAsync() throws Exception {
266-
String topic = "persistent://my-property/my-ns/testProducerDeduplicationNonBatchAsync";
267-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
251+
String topic = newTopicName();
252+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
268253

269254
// Set infinite timeout
270255
ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topic)
@@ -295,8 +280,8 @@ public void testProducerDeduplicationNonBatchAsync() throws Exception {
295280
Message<byte[]> msg = consumer.receive(1, TimeUnit.SECONDS);
296281
assertNull(msg);
297282

298-
// Kill and restart broker
299-
restartBroker();
283+
// Unload topic to force reload of dedup state
284+
admin.topics().unload(topic);
300285

301286
producer = producerBuilder.create();
302287
assertEquals(producer.getLastSequenceId(), 5L);
@@ -313,8 +298,8 @@ public void testProducerDeduplicationNonBatchAsync() throws Exception {
313298

314299
@Test(timeOut = 30000)
315300
public void testKeyBasedBatchingOrder() throws Exception {
316-
final String topic = "persistent://my-property/my-ns/test-key-based-batching-order";
317-
admin.namespaces().setDeduplicationStatus("my-property/my-ns", true);
301+
final String topic = newTopicName();
302+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
318303

319304
final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
320305
.topic(topic)
@@ -379,13 +364,13 @@ public void testKeyBasedBatchingOrder() throws Exception {
379364

380365
@Test
381366
public void testUpdateSequenceIdInSyncCodeSegment() throws Exception {
382-
final String topic = "persistent://my-property/my-ns/testUpdateSequenceIdInSyncCodeSegment";
367+
final String topic = newTopicName();
383368
int totalMessage = 200;
384369
int threadSize = 5;
385-
String topicName = "subscription";
370+
String subscriptionName = "subscription";
386371
@Cleanup("shutdownNow")
387372
ExecutorService executorService = Executors.newFixedThreadPool(threadSize);
388-
conf.setBrokerDeduplicationEnabled(true);
373+
admin.namespaces().setDeduplicationStatus(getNamespace(), true);
389374

390375
//build producer/consumer
391376
Producer<byte[]> producer = pulsarClient.newProducer()
@@ -397,7 +382,7 @@ public void testUpdateSequenceIdInSyncCodeSegment() throws Exception {
397382
Consumer<byte[]> consumer = pulsarClient.newConsumer()
398383
.topic(topic)
399384
.subscriptionType(SubscriptionType.Exclusive)
400-
.subscriptionName(topicName)
385+
.subscriptionName(subscriptionName)
401386
.subscribe();
402387

403388
CountDownLatch countDownLatch = new CountDownLatch(threadSize);

0 commit comments

Comments
 (0)