test: add tests for lazy-loading

apache · Apr 21, 2021 · d49aab4 · d49aab4
1 parent a04a7be
commit d49aab4
Show file tree

Hide file tree

Showing 12 changed files with 482 additions and 21 deletions.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -45,6 +45,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import javax.ws.rs.core.Response.Status;
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
@@ -1521,6 +1522,17 @@ public void testMaxTopicsPerNamespace() throws Exception {
         }
 
         // check producer/consumer auto create partitioned topic
+        final Function<Producer<byte[]>, Producer<byte[]>> send = (p) -> {
+            for (int i = 0; i < 3; i++) {
+                try {
+                    p.newMessage().value("msg".getBytes()).send();
+                } catch (Throwable e) {
+                    log.info("Exception: ", e);
+                    fail();
+                }
+            }
+            return p;
+        };
         super.internalCleanup();
         conf.setMaxTopicsPerNamespace(10);
         conf.setDefaultNumPartitions(3);
@@ -1530,8 +1542,8 @@ public void testMaxTopicsPerNamespace() throws Exception {
         admin.tenants().createTenant("testTenant", tenantInfo);
         admin.namespaces().createNamespace("testTenant/ns1", Sets.newHashSet("test"));
 
-        pulsarClient.newProducer().topic(topic + "1").create().close();
-        pulsarClient.newProducer().topic(topic + "2").create().close();
+        send.apply(pulsarClient.newProducer().topic(topic + "1").enableBatching(false).create()).close();
+        send.apply(pulsarClient.newProducer().topic(topic + "2").enableBatching(false).create()).close();
         pulsarClient.newConsumer().topic(topic + "3").subscriptionName("test_sub").subscribe().close();
         try {
             pulsarClient.newConsumer().topic(topic + "4").subscriptionName("test_sub").subscribe().close();

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -48,6 +48,7 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
 import javax.ws.rs.BadRequestException;
 import javax.ws.rs.ClientErrorException;
 import javax.ws.rs.WebApplicationException;
@@ -1533,6 +1534,17 @@ public void testMaxTopicsPerNamespace() throws Exception {
 
 
         // check producer/consumer auto create partitioned topic
+        final Function<Producer<byte[]>, Producer<byte[]>> send = (p) -> {
+            for (int i = 0; i < 3; i++) {
+                try {
+                    p.newMessage().value("msg".getBytes()).send();
+                } catch (Throwable e) {
+                    log.info("Exception: ", e);
+                    fail();
+                }
+            }
+            return p;
+        };
         cleanup();
         conf.setMaxTopicsPerNamespace(0);
         conf.setDefaultNumPartitions(3);
@@ -1544,8 +1556,8 @@ public void testMaxTopicsPerNamespace() throws Exception {
         admin.namespaces().createNamespace(namespace, Sets.newHashSet("use"));
         admin.namespaces().setMaxTopicsPerNamespace(namespace, 10);
 
-        pulsarClient.newProducer().topic(topic + "1").create().close();
-        pulsarClient.newProducer().topic(topic + "2").create().close();
+        send.apply(pulsarClient.newProducer().topic(topic + "1").enableBatching(false).create()).close();
+        send.apply(pulsarClient.newProducer().topic(topic + "2").enableBatching(false).create()).close();
         pulsarClient.newConsumer().topic(topic + "3").subscriptionName("test_sub").subscribe().close();
 
         try {
@@ -1558,7 +1570,7 @@ public void testMaxTopicsPerNamespace() throws Exception {
         // remove namespace limit
         admin.namespaces().removeMaxTopicsPerNamespace(namespace);
         for (int i = 0; i < 10; ++i) {
-            pulsarClient.newProducer().topic(topic + "_p" + i).create().close();
+            send.apply(pulsarClient.newProducer().topic(topic + "_p" + i).enableBatching(false).create()).close();
             pulsarClient.newConsumer().topic(topic + "_c" + i).subscriptionName("test_sub").subscribe().close();
         }
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java
@@ -20,6 +20,8 @@
 package org.apache.pulsar.broker.admin;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
 import java.util.List;
 import java.util.UUID;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -57,8 +59,17 @@ public void testPartitionedTopicAutoCreation() throws PulsarAdminException, Puls
 
         Producer<byte[]> producer = pulsarClient.newProducer()
                 .topic(topic)
+                .enableBatching(false)
                 .create();
 
+        for (int i = 0; i < 3; i++) {
+            try {
+                producer.newMessage().value("msg".getBytes()).send();
+            } catch (Throwable e) {
+                fail();
+            }
+        }
+
         List<String> partitionedTopics = admin.topics().getPartitionedTopicList(namespaceName);
         List<String> topics = admin.topics().getList(namespaceName);
         assertEquals(partitionedTopics.size(), 1);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -68,6 +68,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
@@ -685,6 +686,17 @@ public void testGetMaxProducerApplied() throws Exception {
     public void testSetMaxProducers() throws Exception {
         Integer maxProducers = 2;
         log.info("MaxProducers: {} will set to the topic: {}", maxProducers, persistenceTopic);
+        final Function<Producer<byte[]>, Producer<byte[]>> send = (p) -> {
+            for (int i = 0; i < 2; i++) {
+                try {
+                    p.newMessage().value("msg".getBytes()).send();
+                } catch (Throwable e) {
+                    log.info("Exception: ", e);
+                    fail();
+                }
+            }
+            return p;
+        };
 
         Awaitility.await().atMost(3, TimeUnit.SECONDS)
                 .until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
@@ -695,8 +707,8 @@ public void testSetMaxProducers() throws Exception {
                 .untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxProducers(persistenceTopic), maxProducers));
 
         admin.topics().createPartitionedTopic(persistenceTopic, 2);
-        Producer<byte[]> producer1 = pulsarClient.newProducer().topic(persistenceTopic).create();
-        Producer<byte[]> producer2 = pulsarClient.newProducer().topic(persistenceTopic).create();
+        Producer<byte[]> producer1 = send.apply(pulsarClient.newProducer().topic(persistenceTopic).enableBatching(false).create());
+        Producer<byte[]> producer2 = send.apply(pulsarClient.newProducer().topic(persistenceTopic).enableBatching(false).create());
         Producer<byte[]> producer3 = null;
 
         try {
@@ -717,6 +729,17 @@ public void testSetMaxProducers() throws Exception {
     public void testRemoveMaxProducers() throws Exception {
         Integer maxProducers = 2;
         log.info("MaxProducers: {} will set to the topic: {}", maxProducers, persistenceTopic);
+        final Function<Producer<byte[]>, Producer<byte[]>> send = (p) -> {
+            for (int i = 0; i < 2; i++) {
+                try {
+                    p.newMessage().value("msg".getBytes()).send();
+                } catch (Throwable e) {
+                    log.info("Exception: ", e);
+                    fail();
+                }
+            }
+            return p;
+        };
 
         Awaitility.await().atMost(3, TimeUnit.SECONDS)
                 .until(() -> pulsar.getTopicPoliciesService().cacheIsInitialized(TopicName.get(testTopic)));
@@ -727,8 +750,8 @@ public void testRemoveMaxProducers() throws Exception {
                 .untilAsserted(() -> Assert.assertEquals(admin.topics().getMaxProducers(persistenceTopic), maxProducers));
 
         admin.topics().createPartitionedTopic(persistenceTopic, 2);
-        Producer<byte[]> producer1 = pulsarClient.newProducer().topic(persistenceTopic).create();
-        Producer<byte[]> producer2 = pulsarClient.newProducer().topic(persistenceTopic).create();
+        Producer<byte[]> producer1 = send.apply(pulsarClient.newProducer().topic(persistenceTopic).enableBatching(false).create());
+        Producer<byte[]> producer2 = send.apply(pulsarClient.newProducer().topic(persistenceTopic).enableBatching(false).create());
         Producer<byte[]> producer3 = null;
         Producer<byte[]> producer4 = null;
 
@@ -747,7 +770,7 @@ public void testRemoveMaxProducers() throws Exception {
         Awaitility.await().atMost(3, TimeUnit.SECONDS)
                 .untilAsserted(() -> Assert.assertNull(admin.topics().getMaxProducers(persistenceTopic)));
 
-        producer3 = pulsarClient.newProducer().topic(persistenceTopic).create();
+        producer3 = send.apply(pulsarClient.newProducer().topic(persistenceTopic).enableBatching(false).create());
         Assert.assertNotNull(producer3);
         admin.namespaces().setMaxProducersPerTopic(myNamespace, 3);
 

diff --git a/...er/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java b/...er/src/test/java/org/apache/pulsar/broker/service/BrokerServiceAutoTopicCreationTest.java
@@ -18,12 +18,15 @@
  */
 package org.apache.pulsar.broker.service;
 
+import java.util.function.Function;
+
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
@@ -101,7 +104,14 @@ public void testAutoPartitionedTopicCreationOnProduce() throws Exception{
         pulsar.getConfiguration().setDefaultNumPartitions(3);
 
         final String topicString = "persistent://prop/ns-abc/partitioned-topic-1";
-        pulsarClient.newProducer().topic(topicString).create();
+        final Producer<byte[]> producer = pulsarClient.newProducer().topic(topicString).enableBatching(false).create();
+        for (int i = 0; i < 3; i++) {
+            try {
+                producer.newMessage().value("msg".getBytes()).send();
+            } catch (Throwable e) {
+                fail();
+            }
+        }
 
         assertTrue(admin.topics().getPartitionedTopicList("prop/ns-abc").contains(topicString));
         for (int i = 0; i < 3; i++) {

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java
@@ -417,7 +417,11 @@ public void testSeekTimeOnPartitionedTopic() throws Exception {
         long resetTimeInMillis = TimeUnit.SECONDS
                 .toMillis(RelativeTimeUtil.parseRelativeTimeInSeconds(resetTimeStr));
         admin.topics().createPartitionedTopic(topicName, partitions);
-        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).enableBatching(false).create();
+        for (int i = 0; i < 2; i++) {
+            String message = "my-message-" + i;
+            producer.send(message.getBytes());
+        }
         // Disable pre-fetch in consumer to track the messages received
         org.apache.pulsar.client.api.Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
                 .subscriptionName("my-subscription").subscribe();
@@ -463,7 +467,7 @@ public void testSeekTimeOnPartitionedTopic() throws Exception {
         for (PersistentSubscription sub : subs) {
             backlogs += sub.getNumberOfEntriesInBacklog(false);
         }
-        assertEquals(backlogs, 10);
+        assertEquals(backlogs, 12);
     }
 
     @Test

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
@@ -33,6 +33,7 @@
 
 import lombok.Cleanup;
 import org.apache.pulsar.client.impl.ConsumerBase;
+import org.apache.pulsar.client.impl.PartitionedProducerImpl;
 import org.apache.pulsar.client.impl.ProducerBase;
 import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse.LookupType;
 import org.apache.pulsar.common.api.proto.ServerError;
@@ -462,8 +463,90 @@ private void subscribeFailDoesNotFailOtherConsumer(String topic1, String topic2)
         mockBrokerService.resetHandleSubscribe();
     }
 
-    // if a producer fails to connect while creating partitioned producer, it should close all successful connections of
-    // other producers and fail
+    // failed to connect to partition at initialization step
+    @Test
+    public void testPartitionedProducerFailOnInitialization() throws Throwable {
+        @Cleanup
+        PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getHttpAddress()).build();
+        final AtomicInteger producerCounter = new AtomicInteger(0);
+
+        mockBrokerService.setHandleProducer((ctx, producer) -> {
+            if (producerCounter.incrementAndGet() == 1) {
+                ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.AuthorizationError, "msg"));
+                return;
+            }
+            ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "default-producer", SchemaVersion.Empty));
+        });
+
+        try {
+            client.newProducer().accessMode(ProducerAccessMode.Shared).topic("persistent://prop/use/ns/multi-part-t1").create();
+            fail("Should have failed with an authorization error");
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarClientException.AuthorizationException);
+        }
+
+        mockBrokerService.resetHandleProducer();
+        mockBrokerService.resetHandleCloseProducer();
+        client.close();
+    }
+
+    // failed to connect to partition at sending step
+    @Test
+    public void testPartitionedProducerFailOnSending() throws Throwable {
+        @Cleanup
+        PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getHttpAddress()).build();
+        final AtomicInteger producerCounter = new AtomicInteger(0);
+        final AtomicInteger closeCounter = new AtomicInteger(0);
+        final String topicName = "persistent://prop/use/ns/multi-part-t1";
+
+        mockBrokerService.setHandleProducer((ctx, producer) -> {
+            if (producerCounter.incrementAndGet() == 2) {
+                ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.AuthorizationError, "msg"));
+                return;
+            }
+            ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "default-producer", SchemaVersion.Empty));
+        });
+
+        mockBrokerService.setHandleSend((ctx, send, headersAndPayload) ->
+            ctx.writeAndFlush(Commands.newSendReceipt(send.getProducerId(), send.getSequenceId(), send.getHighestSequenceId(), 0L, 0L))
+        );
+
+        mockBrokerService.setHandleCloseProducer((ctx, closeProducer) -> {
+            ctx.writeAndFlush(Commands.newSuccess(closeProducer.getRequestId()));
+            closeCounter.incrementAndGet();
+        });
+
+        final PartitionedProducerImpl<byte[]> producer = (PartitionedProducerImpl<byte[]>) client.newProducer()
+                .accessMode(ProducerAccessMode.Shared).topic(topicName).enableBatching(false).create();
+
+        try {
+            producer.send("msg".getBytes());
+            fail("Should have failed with an not connected exception");
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarClientException.NotConnectedException);
+            assertEquals(producer.getProducers().size(), 1);
+        }
+
+        try {
+            // recreate failed producer
+            for (int i = 0; i < client.getPartitionsForTopic(topicName).get().size(); i++) {
+                producer.send("msg".getBytes());
+            }
+            assertEquals(producer.getProducers().size(), client.getPartitionsForTopic(topicName).get().size());
+        } catch (Exception e) {
+            fail();
+        }
+
+        // should not call close
+        assertEquals(closeCounter.get(), 0);
+
+        mockBrokerService.resetHandleProducer();
+        mockBrokerService.resetHandleCloseProducer();
+        client.close();
+    }
+
+    // if a producer which doesn't connect as Shared mode fails to connect while creating partitioned producer,
+    // it should close all successful connections of other producers and fail
     @Test
     public void testOneProducerFailShouldCloseAllProducersInPartitionedProducer() throws Exception {
         @Cleanup
@@ -485,7 +568,19 @@ public void testOneProducerFailShouldCloseAllProducersInPartitionedProducer() th
         });
 
         try {
-            client.newProducer().topic("persistent://prop/use/ns/multi-part-t1").create();
+            client.newProducer().accessMode(ProducerAccessMode.Exclusive).topic("persistent://prop/use/ns/multi-part-t1-0").create();
+            fail("Should have failed with an authorization error");
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarClientException.AuthorizationException);
+            // should call close for 3 partitions
+            assertEquals(closeCounter.get(), 3);
+        }
+
+        producerCounter.set(0);
+        closeCounter.set(0);
+
+        try {
+            client.newProducer().accessMode(ProducerAccessMode.WaitForExclusive).topic("persistent://prop/use/ns/multi-part-t1-1").create();
             fail("Should have failed with an authorization error");
         } catch (Exception e) {
             assertTrue(e instanceof PulsarClientException.AuthorizationException);

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentTopicTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import static org.apache.pulsar.client.util.MathUtils.signSafeMod;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
@@ -40,6 +41,7 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import lombok.Cleanup;
@@ -145,7 +147,16 @@ public void testAutoCreateNonPersistentPartitionsWhenThePartitionedTopicExists()
             assertEquals(consumer.getConsumers().size(), 3);
 
             // When produce, a sub-producer is created for each partition which means the partitions are created
-            PartitionedProducerImpl<byte[]> producer = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer().topic(topic).create();
+            PartitionedProducerImpl<byte[]> producer = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+                    .enableBatching(false).topic(topic).create();
+            for (int i = 0; i < 3; i++) {
+                try {
+                    producer.newMessage().value("msg".getBytes()).send();
+                } catch (Throwable e) {
+                    log.info("Exception: ", e);
+                    fail();
+                }
+            }
             assertEquals(producer.getProducers().size(), 3);
 
             consumer.close();