[pulsar] branch branch-3.1 updated: [fix][broker] Fix heartbeat namespace create transaction internal topic (#21348)

[technoboy]

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

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


The following commit(s) were added to refs/heads/branch-3.1 by this push:
     new aba4764f11a [fix][broker] Fix heartbeat namespace create transaction 
internal topic (#21348)
aba4764f11a is described below

commit aba4764f11a2673e7d64b7d6148e157cd6de6f56
Author: ken <1647023...@qq.com>
AuthorDate: Thu Oct 19 10:18:12 2023 +0800

    [fix][broker] Fix heartbeat namespace create transaction internal topic 
(#21348)
---
 .../broker/service/persistent/PersistentTopic.java    |  3 ++-
 .../broker/systopic/PartitionedSystemTopicTest.java   | 19 +++++++++++++++++++
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 1bfe55cf74c..a090aa1bf25 100644
--- 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -305,7 +305,8 @@ public class PersistentTopic extends AbstractTopic 
implements Topic, AddEntryCal
 
         TopicName topicName = TopicName.get(topic);
         if 
(brokerService.getPulsar().getConfiguration().isTransactionCoordinatorEnabled()
-                && !isEventSystemTopic(topicName)) {
+                && !isEventSystemTopic(topicName)
+                && 
!NamespaceService.isHeartbeatNamespace(topicName.getNamespaceObject())) {
             this.transactionBuffer = brokerService.getPulsar()
                     .getTransactionBufferProvider().newTransactionBuffer(this);
         } else {
diff --git 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
index 4af0bd90523..42d941e6168 100644
--- 
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
+++ 
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
@@ -81,6 +81,7 @@ public class PartitionedSystemTopicTest extends 
BrokerTestBase {
         conf.setDefaultNumPartitions(PARTITIONS);
         conf.setManagedLedgerMaxEntriesPerLedger(1);
         conf.setBrokerDeleteInactiveTopicsEnabled(false);
+        conf.setTransactionCoordinatorEnabled(true);
 
         super.baseSetup();
     }
@@ -207,6 +208,24 @@ public class PartitionedSystemTopicTest extends 
BrokerTestBase {
         });
     }
 
+    @Test
+    public void testHeartbeatNamespaceNotCreateTransactionInternalTopic() 
throws Exception {
+        admin.brokers().healthcheck(TopicVersion.V2);
+        NamespaceName namespaceName = 
NamespaceService.getHeartbeatNamespaceV2(pulsar.getLookupServiceAddress(),
+                pulsar.getConfig());
+        TopicName topicName = TopicName.get("persistent",
+                namespaceName, SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT);
+        Optional<Topic> optionalTopic = pulsar.getBrokerService()
+                .getTopic(topicName.getPartition(1).toString(), false).join();
+        Assert.assertTrue(optionalTopic.isEmpty());
+
+        List<String> topics = 
getPulsar().getNamespaceService().getListOfPersistentTopics(namespaceName).join();
+        Assert.assertEquals(topics.size(), 1);
+        TopicName heartbeatTopicName = TopicName.get("persistent",
+                namespaceName, BrokersBase.HEALTH_CHECK_TOPIC_SUFFIX);
+        Assert.assertEquals(topics.get(0), heartbeatTopicName.toString());
+    }
+
     @Test
     public void testSetBacklogCausedCreatingProducerFailure() throws Exception 
{
         final String ns = "prop/ns-test";