在上一篇 Pulsar3.0新功能介紹中提到，在升級到 3.0 的過程中碰到一個(gè)致命的問題，就是升級之后 topic 被刪除了。

正好最近社區(qū)也補(bǔ)充了相關(guān)細(xì)節(jié)，本次也接著這個(gè)機(jī)會(huì)再次復(fù)盤一下，畢竟這是一個(gè)非常致命的 Bug。

圖片

現(xiàn)象

先來回顧下當(dāng)時(shí)的情況：升級當(dāng)晚沒有出現(xiàn)啥問題，各個(gè)流量指標(biāo)、生產(chǎn)者、消費(fèi)者數(shù)量都是在正常范圍內(nèi)波動(dòng)。

事后才知道，因?yàn)橹皇莿h除了很少一部分的 topic，所以從監(jiān)控中反應(yīng)不出來。

早上上班后陸續(xù)有部分業(yè)務(wù)反饋應(yīng)用連不上 topic，提示 topic nof found.

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'Producer': Invocation of init method failed; nested exception is org.apache.pulsar.client.api.PulsarClientException$TopicDoesNotExistException: Topic Not Found.

因?yàn)橹皇遣糠謶?yīng)用在反饋，所以起初懷疑是 broker 升級之后導(dǎo)致老版本的 pulsar-client 存在兼容性問題。

所以我就拿了平時(shí)測試用的 topic 再配合多個(gè)老版本的 sdk 進(jìn)行測試，發(fā)現(xiàn)沒有問題。

直到這一步還好，至少證明是小范故障。

因?yàn)樘崾镜氖?topic 不存在，所以就準(zhǔn)備查一下 topic 的元數(shù)據(jù)是否正常。

圖片

查詢后發(fā)現(xiàn)元數(shù)據(jù)是存在的。

之后我便想看看提示了 topic 不存在的 topic 的歸屬，然后再看看那個(gè) broker 中是否有異常日志。

圖片

發(fā)現(xiàn)查看歸屬的接口也是提示 topic 不存在，此時(shí)我便懷疑是 topic 的負(fù)載出現(xiàn)了問題，導(dǎo)致這些 topic 沒有綁定到具體的 broker。

于是便重啟了 broker，結(jié)果依然沒有解決問題。

圖片

之后我們查詢了 topic 的 internal state 發(fā)現(xiàn)元數(shù)據(jù)中會(huì)少一個(gè)分區(qū)。

緊急恢復(fù)

我們嘗試將這個(gè)分區(qū)數(shù)恢復(fù)后，發(fā)現(xiàn)這個(gè) topic 就可以正常連接了。

于是再挑選了幾個(gè)異常的 topic 發(fā)現(xiàn)都是同樣的問題，恢復(fù)分區(qū)數(shù)之后也可以正常連接了。

所以我寫了一個(gè)工具遍歷了所有的 topic，檢測分區(qū)數(shù)是否正常，不正常時(shí)便修復(fù)。

void checkPartition() {  
    String namespace = "tenant/ns";  
    List<String> topicList = pulsarAdmin.topics().getPartitionedTopicList(namespace);  
    for (String topic : topicList) {  
        PartitionedTopicStats stats = pulsarAdmin.topics().getPartitionedStats(topic, true);  
        int partitions = stats.getMetadata().partitions;  
        int size = stats.getPartitions().size();  
        if (partitions != size) {  
            log.info("topic={},partitinotallow={},size={}", topic, partitions, size);  
            pulsarAdmin.topics().updatePartitionedTopic(topic, partitions);  
        }    
 }
}

排查

修復(fù)好所有 topic 之后便開始排查根因，因?yàn)榭吹降氖窃獢?shù)據(jù)不一致所以懷疑是 zk 里的數(shù)據(jù)和 broker 內(nèi)存中的數(shù)據(jù)不同導(dǎo)致的這個(gè)問題。

但我們查看了 zookeeper 中的數(shù)據(jù)發(fā)現(xiàn)一切又是正常的，所以只能轉(zhuǎn)變思路。

之后我們通過有問題的 topic 在日志中找到了一個(gè)關(guān)鍵日志：

圖片

圖片

以及具體的堆棧。

此時(shí)具體的原因已經(jīng)很明顯了，元數(shù)據(jù)這些自然是沒問題；根本原因是 topic 被刪除了，但被刪除的 topic 只是某個(gè)分區(qū)，所以我們在查詢 internalState 時(shí)才發(fā)發(fā)現(xiàn)少一個(gè) topic。

通過這個(gè)刪除日志定位到具體的刪除代碼：

org.apache.pulsar.broker.service.persistent.PersistentTopic#checkReplication

圖片

原來是這里的  configuredClusters 值為空才導(dǎo)致的 topic 調(diào)用了 deleteForcefully()被刪除。

而這個(gè)值是從 topic 的 Policy 中獲取的。

復(fù)現(xiàn)問題

通過上圖中的堆棧跟蹤，懷疑是重啟  broker 導(dǎo)致的 topic unload ，同時(shí) broker 又在構(gòu)建 topic 導(dǎo)致了對 topicPolicy 的讀寫。

最終導(dǎo)致 topicPolicy 為空。

只要寫個(gè)單測可以復(fù)現(xiàn)這個(gè)問題就好辦了：

@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);
        }

    }

同時(shí)還得查詢元數(shù)據(jù)有耗時(shí)才能復(fù)現(xiàn)：

圖片

只能手動(dòng) sleep 模擬這個(gè)耗時(shí)

具體也可以參考這個(gè) issue https://github.com/apache/pulsar/issues/21653#issuecomment-1842962452

此時(shí)就會(huì)發(fā)現(xiàn)有 topic 被刪除了，而且是隨機(jī)刪除的，因?yàn)槌霈F(xiàn)并發(fā)的幾率本身也是隨機(jī)的。

圖片

這里畫了一個(gè)流程圖就比較清晰了，在 broker 重啟的時(shí)候會(huì)有兩個(gè)線程同時(shí)topicPolicy 進(jìn)行操作。

在 thread3 讀取 topicPolicy 進(jìn)行判斷時(shí)，thread2 可能還沒有把數(shù)據(jù)準(zhǔn)備好，所以就導(dǎo)致了 topic 被刪除。

修復(fù)

既然知道了問題原因就好修復(fù)了，我們只需要把 thread3 和 thread2 修改為串行執(zhí)行就好了。

這也是處理并發(fā)最簡單高效的方法，就是直接避免并發(fā)；加鎖、隊(duì)列啥的雖然也可以解決，但代碼復(fù)雜度也高了很多，所以能不并發(fā)就盡量不要并發(fā)。

但要把這個(gè)修復(fù)推送到社區(qū)上游主分支最好是要加上單測，這樣即便是后續(xù)有其他的改動(dòng)也能保證這個(gè) bug 不會(huì)再次出現(xiàn)。

之后在社區(qū)大佬的幫助下完善了單測，最終合并了這個(gè)修復(fù)。

再次證明寫單測往往比代碼更復(fù)雜，也更花費(fèi)時(shí)間。

PR：https://github.com/apache/pulsar/pull/21704

使用修復(fù)鏡像

因?yàn)樯鐓^(qū)合并代碼再發(fā)版的周期較長，而我們又急于修復(fù)該問題；不然都不敢重啟 broker，因?yàn)槊恐貑⒁淮味伎赡軙?huì)導(dǎo)致不知道哪個(gè) topic 就被刪除了。

所以我們自己在本地構(gòu)建了一個(gè)修復(fù)的鏡像，準(zhǔn)備在線上進(jìn)行替換。

此時(shí)坑又來了，我們滿懷信心的替換了一個(gè)鏡像再觀察日志發(fā)現(xiàn)居然還有刪除的日志??。

冷靜下來一分析，原來是當(dāng)前替換進(jìn)行的 broker 沒有問題了，但它處理的 topic 被轉(zhuǎn)移到了其他 broker 中，而其他的 broker 并沒有替換為我們最新的鏡像。

所以導(dǎo)致 topic 在其他 broker 中依然被刪除了。

除非我們停機(jī)，將所有的鏡像都替換之后再一起重啟。

但這樣的成本太高了，最好是可以平滑發(fā)布。

最終我們想到一個(gè)辦法，使用 arthas 去關(guān)閉了一個(gè) broker 的一個(gè)選項(xiàng)，之后就不會(huì)執(zhí)行出現(xiàn) bug 的那段代碼了。

圖片

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)'"

我也將操作方法貼到了對于 issue 的評論區(qū)。https://github.com/apache/pulsar/issues/21653#issuecomment-1857548997如果不幸碰到了這個(gè) bug，可以參考修復(fù)。

總結(jié)

刪除的這些 topic 的同時(shí)它的訂閱者也被刪除了，所以我們還需要修復(fù)訂閱者：

String topicName = "persistent://tenant/ns/topicName";  
pulsarTopicService.createSubscription(topicName, "subName", MessageId.latest);

之所以說這個(gè) bug 非常致命，是因?yàn)檫@樣會(huì)導(dǎo)致 topic 的數(shù)據(jù)丟失，同時(shí)這些 topic 上的數(shù)據(jù)也會(huì)被刪除。

后續(xù) https://github.com/apache/pulsar/pull/21704#issuecomment-1878315926社區(qū)也補(bǔ)充了一些場景。

圖片

其實(shí)場景 2 更容易出現(xiàn)復(fù)現(xiàn)，畢竟更容易出現(xiàn)并發(fā)；也就是我們碰到的場景

說來也奇怪，結(jié)合社區(qū)的 issue 和其他大佬的反饋，這個(gè)問題只有我們碰到了，估計(jì)也是這個(gè)問題的觸發(fā)條件也比較苛刻：

• 開啟 systemTopic/topicLevelPolicessystemTopicEnabled: "true" topicLevelPoliciesEnabled: "true"
• 設(shè)置足夠多的 topicPolicies
• 重啟 broker

重啟過程中從 zk 中獲取數(shù)據(jù)出現(xiàn)耗時(shí)

符合以上條件的集群就需要注意了。

其實(shí)這個(gè)問題在這個(gè) PR 就已經(jīng)引入了 https://github.com/apache/pulsar/pull/11021

所以已經(jīng)存在蠻久了，后續(xù)我們也將檢測元數(shù)據(jù)作為升級流程之一了，確保升級后數(shù)據(jù)依然是完整的。

相關(guān)的 issue 和 PR： 

https://github.com/apache/pulsar/issues/21653 https://github.com/apache/pulsar/pull/21704