This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 6ec15d88e99 [improve] [broker] [break change] Do not create
partitioned DLQ/Retry topic automatically (#22705)
6ec15d88e99 is described below
commit 6ec15d88e99a640f74777d09a1a1ff2945dff517
Author: fengyubiao <yubiao.f...@streamnative.io>
AuthorDate: Thu May 16 02:47:26 2024 +0800
[improve] [broker] [break change] Do not create partitioned DLQ/Retry topic
automatically (#22705)
(cherry picked from commit f07b3a030179c38f9786b3e26c82aa13e00b34a6)
---
.../pulsar/broker/service/BrokerService.java | 6 +
.../DeadLetterTopicDefaultMultiPartitionsTest.java | 251 +++++++++++++++++++++
2 files changed, 257 insertions(+)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b74e6ee2c4e..85930f06856 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -23,6 +23,8 @@ import static java.util.Objects.requireNonNull;
import static
org.apache.bookkeeper.mledger.ManagedLedgerConfig.PROPERTY_SOURCE_TOPIC_KEY;
import static org.apache.commons.collections4.CollectionUtils.isEmpty;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static
org.apache.pulsar.client.util.RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
+import static
org.apache.pulsar.client.util.RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
import static
org.apache.pulsar.common.naming.SystemTopicNames.isTransactionInternalName;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Queues;
@@ -3525,6 +3527,10 @@ public class BrokerService implements Closeable {
}
public boolean isDefaultTopicTypePartitioned(final TopicName topicName,
final Optional<Policies> policies) {
+ if
(topicName.getPartitionedTopicName().endsWith(DLQ_GROUP_TOPIC_SUFFIX)
+ ||
topicName.getPartitionedTopicName().endsWith(RETRY_GROUP_TOPIC_SUFFIX)) {
+ return false;
+ }
AutoTopicCreationOverride autoTopicCreationOverride =
getAutoTopicCreationOverride(topicName, policies);
if (autoTopicCreationOverride != null) {
return
TopicType.PARTITIONED.toString().equals(autoTopicCreationOverride.getTopicType());
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicDefaultMultiPartitionsTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicDefaultMultiPartitionsTest.java
new file mode 100644
index 00000000000..b8bccb79372
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicDefaultMultiPartitionsTest.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import static
org.apache.pulsar.client.util.RetryMessageUtil.DLQ_GROUP_TOPIC_SUFFIX;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.TopicType;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker-impl")
+public class DeadLetterTopicDefaultMultiPartitionsTest extends
ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ this.conf.setMaxMessageSize(5 * 1024);
+ this.conf.setAllowAutoTopicCreation(true);
+ this.conf.setDefaultNumPartitions(2);
+ this.conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED);
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ private void triggerDLQGenerate(String topic, String subscription) throws
Exception {
+ String DLQ = getDLQName(topic, subscription);
+ String p0OfDLQ = TopicName.get(DLQ).getPartition(0).toString();
+ Consumer consumer =
pulsarClient.newConsumer().topic(topic).subscriptionName(subscription)
+ .ackTimeout(1000, TimeUnit.MILLISECONDS)
+ .subscriptionType(SubscriptionType.Shared)
+ .receiverQueueSize(10)
+ .negativeAckRedeliveryDelay(100, TimeUnit.MILLISECONDS)
+
.deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(1).build())
+ .subscribe();
+ Producer producer = pulsarClient.newProducer().topic(topic).create();
+ producer.newMessage().value(new byte[]{1}).send();
+
+ Message<Integer> message1 = consumer.receive();
+ consumer.negativeAcknowledge(message1);
+ Message<Integer> message2 = consumer.receive();
+ consumer.negativeAcknowledge(message2);
+
+ Awaitility.await().atMost(Duration.ofSeconds(1500)).until(() -> {
+ Message<Integer> message3 = consumer.receive(2, TimeUnit.SECONDS);
+ if (message3 != null) {
+ log.info("===> {}", message3.getRedeliveryCount());
+ consumer.negativeAcknowledge(message3);
+ }
+ List<String> topicList =
pulsar.getPulsarResources().getTopicResources()
+
.listPersistentTopicsAsync(TopicName.get(topic).getNamespaceObject()).join();
+ if (topicList.contains(DLQ) || topicList.contains(p0OfDLQ)) {
+ return true;
+ }
+ int partitions =
admin.topics().getPartitionedTopicMetadata(topic).partitions;
+ for (int i = 0; i < partitions; i++) {
+ for (int j = -1; j <
pulsar.getConfig().getDefaultNumPartitions(); j++) {
+ String p0OfDLQ2;
+ if (j == -1) {
+ p0OfDLQ2 = TopicName
+
.get(getDLQName(TopicName.get(topic).getPartition(i).toString(), subscription))
+ .toString();
+ } else {
+ p0OfDLQ2 = TopicName
+
.get(getDLQName(TopicName.get(topic).getPartition(i).toString(), subscription))
+ .getPartition(j).toString();
+ }
+ if (topicList.contains(p0OfDLQ2)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ });
+ producer.close();
+ consumer.close();
+ admin.topics().unload(topic);
+ }
+
+ private static String getDLQName(String primaryTopic, String subscription)
{
+ String domain = TopicName.get(primaryTopic).getDomain().toString();
+ return domain + "://" + TopicName.get(primaryTopic)
+ .toString().substring(( domain + "://").length())
+ + "-" + subscription + DLQ_GROUP_TOPIC_SUFFIX;
+ }
+
+ @DataProvider(name = "topicCreationTypes")
+ public Object[][] topicCreationTypes() {
+ return new Object[][]{
+ //{TopicType.NON_PARTITIONED},
+ {TopicType.PARTITIONED}
+ };
+ }
+
+ @Test(dataProvider = "topicCreationTypes")
+ public void testGenerateNonPartitionedDLQ(TopicType topicType) throws
Exception {
+ final String topic = BrokerTestUtil.newUniqueName(
"persistent://public/default/tp");
+ final String subscription = "s1";
+ switch (topicType) {
+ case PARTITIONED: {
+ admin.topics().createPartitionedTopic(topic, 2);
+ break;
+ }
+ case NON_PARTITIONED: {
+ admin.topics().createNonPartitionedTopic(topic);
+ }
+ }
+
+ triggerDLQGenerate(topic, subscription);
+
+ // Verify: no partitioned DLQ.
+ List<String> partitionedTopics =
pulsar.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources()
+
.listPartitionedTopicsAsync(TopicName.get(topic).getNamespaceObject(),
TopicDomain.persistent).join();
+ for (String tp : partitionedTopics) {
+ assertFalse(tp.endsWith("-DLQ"));
+ }
+ // Verify: non-partitioned DLQ exists.
+ List<String> partitions =
pulsar.getPulsarResources().getTopicResources()
+
.listPersistentTopicsAsync(TopicName.get(topic).getNamespaceObject()).join();
+ List<String> DLQCreated = new ArrayList<>();
+ for (String tp : partitions) {
+ if (tp.endsWith("-DLQ")) {
+ DLQCreated.add(tp);
+ }
+ assertFalse(tp.endsWith("-partition-0-DLQ"));
+ }
+ assertTrue(!DLQCreated.isEmpty());
+
+ // cleanup.
+ switch (topicType) {
+ case PARTITIONED: {
+ admin.topics().deletePartitionedTopic(topic);
+ break;
+ }
+ case NON_PARTITIONED: {
+ admin.topics().delete(topic, false);
+ }
+ }
+ for (String t : DLQCreated) {
+ try {
+
admin.topics().delete(TopicName.get(t).getPartitionedTopicName(), false);
+ } catch (Exception ex) {}
+ try {
+
admin.topics().deletePartitionedTopic(TopicName.get(t).getPartitionedTopicName(),
false);
+ } catch (Exception ex) {}
+ }
+ }
+
+ @Test
+ public void testManuallyCreatePartitionedDLQ() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName(
"persistent://public/default/tp");
+ final String subscription = "s1";
+ String DLQ = getDLQName(topic, subscription);
+ String p0OfDLQ = TopicName.get(DLQ).getPartition(0).toString();
+ String p1OfDLQ = TopicName.get(DLQ).getPartition(1).toString();
+ admin.topics().createNonPartitionedTopic(topic);
+ admin.topics().createPartitionedTopic(DLQ, 2);
+
+ Awaitility.await().untilAsserted(() -> {
+ // Verify: partitioned DLQ exists.
+ List<String> partitionedTopics =
pulsar.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources()
+
.listPartitionedTopicsAsync(TopicName.get(topic).getNamespaceObject(),
TopicDomain.persistent).join();
+ assertTrue(partitionedTopics.contains(DLQ));
+ assertFalse(partitionedTopics.contains(p0OfDLQ));
+ // Verify: DLQ partitions exists.
+ List<String> partitions =
pulsar.getPulsarResources().getTopicResources()
+
.listPersistentTopicsAsync(TopicName.get(topic).getNamespaceObject()).join();
+ assertFalse(partitions.contains(DLQ));
+ assertTrue(partitions.contains(p0OfDLQ));
+ assertTrue(partitions.contains(p1OfDLQ));
+ });
+
+ // cleanup.
+ admin.topics().delete(topic, false);
+ admin.topics().deletePartitionedTopic(DLQ, false);
+ }
+
+ @Test
+ public void testManuallyCreatePartitionedDLQ2() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName(
"persistent://public/default/tp");
+ final String subscription = "s1";
+ final String p0OfTopic =
TopicName.get(topic).getPartition(0).toString();
+ String DLQ = getDLQName(p0OfTopic, subscription);
+ String p0OfDLQ = TopicName.get(DLQ).getPartition(0).toString();
+ admin.topics().createPartitionedTopic(topic, 10);
+ try {
+ admin.topics().createPartitionedTopic(DLQ, 2);
+ } catch (Exception ex) {
+ // Keep multiple versions compatible.
+ if (ex.getMessage().contains("Partitioned Topic Name should not
contain '-partition-'")){
+ return;
+ } else {
+ fail("Failed to create partitioned DLQ");
+ }
+ }
+
+ Awaitility.await().untilAsserted(() -> {
+ // Verify: partitioned DLQ exists.
+ List<String> partitionedTopics =
pulsar.getPulsarResources().getNamespaceResources()
+ .getPartitionedTopicResources()
+
.listPartitionedTopicsAsync(TopicName.get(topic).getNamespaceObject(),
TopicDomain.persistent).join();
+ assertTrue(partitionedTopics.contains(DLQ));
+ assertFalse(partitionedTopics.contains(p0OfDLQ));
+ // Verify: DLQ partitions exists.
+ List<String> partitions =
pulsar.getPulsarResources().getTopicResources()
+
.listPersistentTopicsAsync(TopicName.get(topic).getNamespaceObject()).join();
+ assertFalse(partitions.contains(DLQ));
+ });
+
+ // cleanup.
+ admin.topics().deletePartitionedTopic(topic, false);
+ admin.topics().deletePartitionedTopic(DLQ, false);
+ }
+}
