(pulsar) 01/04: [improve][test]Add test: test/testTopicPartitionCannotBeCreatedAfterTopicDeleted (#26038)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 6c68d9c61f6b4f49bc8854cf11a8b9fdd07d3a64
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Jun 19 02:41:53 2026 +0800

    [improve][test]Add test: test/testTopicPartitionCannotBeCreatedAfterTopicDeleted (#26038)
    
    (cherry picked from commit 8eaf5517ff788d0703f75f5deb45ac5b8edb0635)
---
 .../client/api/SimpleProducerConsumerTest.java     | 60 ++++++++++++++++++++++
 1 file changed, 60 insertions(+)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index d5b10fb8347f..dcafd157a908 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -110,6 +110,7 @@ import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
 import org.apache.pulsar.client.impl.PartitionedProducerImpl;
 import org.apache.pulsar.client.impl.ProducerBase;
 import org.apache.pulsar.client.impl.ProducerImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.TopicMessageImpl;
 import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
 import org.apache.pulsar.client.impl.crypto.MessageCryptoBc;
@@ -5438,6 +5439,65 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
         admin.topics().delete(topic, false);
     }
 
+    @Test
+    public void testTopicPartitionCannotBeCreatedAfterTopicDeleted() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+        admin.topics().createPartitionedTopic(topic, 1);
+
+        // Inject an delay: delay to handle channel inactive event, to let the producer delay to reconnect.
+        ClientBuilderImpl clientBuilder = (ClientBuilderImpl) PulsarClient.builder()
+                .serviceUrl(lookupUrl.toString())
+                .statsInterval(0, TimeUnit.SECONDS)
+                .connectionsPerBroker(1);
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        PulsarClientImpl pulsarClient = InjectedClientCnxClientBuilder.create(clientBuilder, (conf, eventLoopGroup) -> {
+
+            return new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup) {
+
+                @Override
+                public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+                    // Delay receiving the event, let producer will not reconnect immediately.
+                    log.info("channel inactive");
+                    countDownLatch.await();
+                    super.channelInactive(ctx);
+                }
+            };
+        });
+
+        // Producer connected.
+        Producer producer = pulsarClient.newProducer(Schema.BYTES)
+                .topic(topic)
+                .create();
+        PersistentTopic persistentTopic1 = (PersistentTopic) pulsar.getBrokerService()
+                .getTopic(topic + "-partition-0", false)
+                .get(5, TimeUnit.SECONDS).get();
+        Awaitility.await().untilAsserted(() -> {
+            Assert.assertEquals(persistentTopic1.getProducers().values().size(), 1);
+        });
+
+        // Make a network issue which leads to the connection breaks.
+        org.apache.pulsar.broker.service.Producer serviceProducer = persistentTopic1.getProducers().values()
+                .iterator().next();
+        ServerCnx servercnx = (ServerCnx) serviceProducer.getCnx();
+        servercnx.ctx().close();
+        // After the connection is break, and before the producer reconnects, the partitioned topic can be deleted
+        // without "--force".
+        admin.topics().deletePartitionedTopic(topic);
+        Awaitility.await().untilAsserted(() -> {
+            assertTrue(persistentTopic1.isClosingOrDeleting());
+        });
+
+        // Verify: the partition can not be loaded up once the partitioned topic was deleted.
+        countDownLatch.countDown();
+        Thread.sleep(10_000);
+        assertFalse(producer.isConnected());
+        assertFalse(pulsar.getBrokerService().getTopics().containsKey(topic + "-partition-0"));
+
+        // cleanup.
+        producer.close();
+        pulsarClient.close();
+    }
+
     /**
      * The internal producer of replicator will resend messages after reconnected. This test guarantees that the
      * internal producer will continuously resent messages even though the client side encounters the following bugs.