[Bug] After upgrading to the 3.0.1, the topic was deleted!

[crossoverJie]

Search before asking

• I searched in the issues and found nothing similar.

Version

3.0.1

Minimal reproduce step

1. broker.conf:

    systemTopicEnabled: "true"
    topicLevelPoliciesEnabled: "true"

3.Setting the publishRate is to ensure the system topic has data

        String namespace = "xxx/xxx";
        List<String> topicList = pulsarAdmin.topics().getPartitionedTopicList(namespace);
        topicList = topicList.subList(0, 20);
        for (String topic : topicList) {
            PublishRate publishRate = new PublishRate();
            publishRate.publishThrottlingRateInMsg = 100;
            pulsarAdmin.topicPolicies().setPublishRate(topic, publishRate);
        }

4. Concurrent query lookup:

            topicList = pulsarAdmin.topics().getPartitionedTopicList(namespace);
            topicList = topicList.subList(0, 20);
            for (String topic : topicList) {
                customThreadPool.execute(()->{
                    try {
                        while (true){
                            pulsarAdmin.lookups().lookupPartitionedTopic(topic);
                        }
                    } catch (PulsarAdminException e) {
                        throw new RuntimeException(e);
                    }
                });
            }

6. Upgrade 3.0.1 from 2.11.1, this will restart broker(Or unload namespace/topic).

If version 3.0.1 has already been installed, It can also be restart broker:

k rollout restart statefulset/pulsar-broker

Eventually it was discovered that some topic partitions had been deleted.

                try {
                    stats = pulsarAdmin.topics().getPartitionedStats(topic, true);
                } catch (PulsarAdminException e) {
                    throw new RuntimeException(e);
                }
                int partitions = stats.getMetadata().partitions;
                int size = stats.getPartitions().size();
                if (partitions != size) {
                    log.info("**********topic={},partitions={},size={}", topic, partitions, size);
//                pulsarAdmin.topics().updatePartitionedTopic(topic, partitions);
                }

What did you expect to see?

Data integrity.

What did you see instead?

About 200+ topics were deleted.

Anything else?

The logs of deleted topics:

The relevant stack logs:

Releted code:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 1569 to 1590 in 90a82ae

	public CompletableFuture<Void> checkReplication() {
	TopicName name = TopicName.get(topic);
	if (!name.isGlobal() || NamespaceService.isHeartbeatNamespace(name)) {
	return CompletableFuture.completedFuture(null);
	}
	if (log.isDebugEnabled()) {
	log.debug("[{}] Checking replication status", name);
	}
	List<String> configuredClusters = topicPolicies.getReplicationClusters().get();
	int newMessageTTLInSeconds = topicPolicies.getMessageTTLInSeconds().get();
	String localCluster = brokerService.pulsar().getConfiguration().getClusterName();
	// if local cluster is removed from global namespace cluster-list : then delete topic forcefully
	// because pulsar doesn't serve global topic without local repl-cluster configured.
	if (TopicName.get(topic).isGlobal() && !configuredClusters.contains(localCluster)) {
	log.info("Deleting topic [{}] because local cluster is not part of "
	+ " global namespace repl list {}", topic, configuredClusters);
	return deleteForcefully();
	}

Are you willing to submit a PR?

• I'm willing to submit a PR!

[crossoverJie]

Reproduce step:

    @Test
    public void testCreateTopicAndUpdatePolicyConcurrent() throws Exception {

        final int topicNum = 100;
        final int partition = 10;

        // (1) Init topic
        admin.namespaces().createNamespace("public/retention");
        final String topicName = "persistent://public/retention/policy_with_broker_restart";
        for (int i = 0; i < topicNum; i++) {
            final String shadowTopicNames = topicName + "_" + i;
            admin.topics().createPartitionedTopic(shadowTopicNames, partition);
        }

        // (2) Set Policy
        for (int i = 90; i < 100; i++) {
            final String shadowTopicNames = topicName + "_" + i;
            CompletableFuture.runAsync(() -> {
                while (true) {
                    PublishRate publishRate = new PublishRate();
                    publishRate.publishThrottlingRateInMsg = 100;
                    try {
                        admin.topicPolicies().setPublishRate(shadowTopicNames, publishRate);
                    } catch (PulsarAdminException e) {
                    }
                }
            });
        }

        for (int i = 90; i < 100; i++) {
            final String shadowTopicNames = topicName + "_" + i;
            CompletableFuture.runAsync(() -> {
                while (true) {
                    try {
                        admin.lookups().lookupPartitionedTopic(shadowTopicNames);
                    } catch (Exception e) {
                    }
                }
            });
        }

        admin.namespaces().unload("public/retention");
        admin.namespaces().unload("public/retention");
        admin.namespaces().unload("public/retention");
        Thread.sleep(1000* 5);

        for (int i = 0; i < topicNum; i++) {
            final String shadowTopicNames = topicName + "_" + i;
            log.info("check topic: {}", shadowTopicNames);
            PartitionedTopicStats partitionedStats = admin.topics().getPartitionedStats(shadowTopicNames, true);
            Assert.assertEquals(partitionedStats.getPartitions().size(), partition);
        }

    }

This issue is caused by the delay in getPoliciesAsync which leads to checkReplication getting an empty array when it tries to get configuredClusters.

We can manually add a delay here to simulate the issue for this unit test to reproduce the problem.

[crossoverJie]

If you unfortunately encounter this bug, please do not immediately try to upgrade to the fixed version.

Because restarting broker may still trigger the deletion of topics.

We handled it this way:

Enter the container where the broker is located, use arthas to modify the condition that triggers this issue, thereby avoiding triggering the deletion logic.

curl -O https://arthas.aliyun.com/arthas-boot.jar && java -jar arthas-boot.jar 1 -c "vmtool -x 3 --action getInstances --className org.apache.pulsar.broker.ServiceConfiguration  --express 'instances[0].setTopicLevelPoliciesEnabled(false)'"

It is important to note that this operation needs to be successfully executed in all brokers before proceeding with the upgrade process.

The operation should be carried out as quickly as possible to avoid potential unknown issues from arising.