[fix][test] Enable testMaxPendingChunkMessages (#18542)

apache · Nov 22, 2022 · 49ff0f8 · 49ff0f8
1 parent 5df8b50
commit 49ff0f8
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java
@@ -309,17 +309,18 @@ public void testPublishWithFailure() throws Exception {
         producer.close();
     }
 
-    @Test(enabled = false)
+    @Test
     public void testMaxPendingChunkMessages() throws Exception {
 
         log.info("-- Starting {} test --", methodName);
-        this.conf.setMaxMessageSize(10);
+        this.conf.setMaxMessageSize(100);
         final int totalMessages = 25;
         final String topicName = "persistent://my-property/my-ns/maxPending";
         final int totalProducers = 25;
         @Cleanup("shutdownNow")
         ExecutorService executor = Executors.newFixedThreadPool(totalProducers);
 
+        @Cleanup
         ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topicName)
                 .subscriptionName("my-subscriber-name").acknowledgmentGroupTime(0, TimeUnit.SECONDS)
                 .maxPendingChunkedMessage(1).autoAckOldestChunkedMessageOnQueueFull(true)
@@ -334,12 +335,11 @@ public void testMaxPendingChunkMessages() throws Exception {
             producers[i] = producerBuilder.enableChunking(true).enableBatching(false).create();
             int index = i;
             executor.submit(() -> {
-                futures.add(producers[index].sendAsync(createMessagePayload(45).getBytes()));
+                futures.add(producers[index].sendAsync(createMessagePayload(450).getBytes()));
             });
         }
 
         FutureUtil.waitForAll(futures).get();
-        PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topicName).get().get();
 
         Message<byte[]> msg = null;
         Set<String> messageSet = new HashSet<>();
@@ -354,8 +354,12 @@ public void testMaxPendingChunkMessages() throws Exception {
             consumer.acknowledge(msg);
         }
 
+        log.info("messageSet size: {}, totalPublishedMessages: {}", messageSet.size(), totalPublishedMessages);
         assertNotEquals(messageSet.size(), totalPublishedMessages);
 
+        for (int i = 0; i < totalProducers; i++) {
+            producers[i].close();
+        }
     }
 
     /**