請注意，你的 Pulsar 集群可能有刪除數據的風險
在上一篇 Pulsar3.0新功能介紹中提到，在升級到 3.0 的過程中碰到一個致命的問題，就是升級之后 topic 被刪除了。
正好最近社區也補充了相關細節，本次也接著這個機會再次復盤一下，畢竟這是一個非常致命的 Bug。
圖片
現象
先來回顧下當時的情況：升級當晚沒有出現啥問題，各個流量指標、生產者、消費者數量都是在正常范圍內波動。
事后才知道，因為只是刪除了很少一部分的 topic，所以從監控中反應不出來。
早上上班后陸續有部分業務反饋應用連不上 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.因為只是部分應用在反饋，所以起初懷疑是 broker 升級之后導致老版本的 pulsar-client 存在兼容性問題。
所以我就拿了平時測試用的 topic 再配合多個老版本的 sdk 進行測試，發現沒有問題。
直到這一步還好，至少證明是小范故障。
因為提示的是 topic 不存在，所以就準備查一下 topic 的元數據是否正常。
圖片
查詢后發現元數據是存在的。
之后我便想看看提示了 topic 不存在的 topic 的歸屬，然后再看看那個 broker 中是否有異常日志。
圖片
發現查看歸屬的接口也是提示 topic 不存在，此時我便懷疑是 topic 的負載出現了問題，導致這些 topic 沒有綁定到具體的 broker。
于是便重啟了 broker，結果依然沒有解決問題。
圖片
之后我們查詢了 topic 的 internal state 發現元數據中會少一個分區。
緊急恢復
我們嘗試將這個分區數恢復后，發現這個 topic 就可以正常連接了。
于是再挑選了幾個異常的 topic 發現都是同樣的問題，恢復分區數之后也可以正常連接了。
所以我寫了一個工具遍歷了所有的 topic，檢測分區數是否正常，不正常時便修復。
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);
}
}
}排查
修復好所有 topic 之后便開始排查根因，因為看到的是元數據不一致所以懷疑是 zk 里的數據和 broker 內存中的數據不同導致的這個問題。
但我們查看了 zookeeper 中的數據發現一切又是正常的，所以只能轉變思路。
之后我們通過有問題的 topic 在日志中找到了一個關鍵日志：
圖片
圖片
以及具體的堆棧。
此時具體的原因已經很明顯了，元數據這些自然是沒問題；根本原因是 topic 被刪除了，但被刪除的 topic 只是某個分區，所以我們在查詢 internalState 時才發發現少一個 topic。
通過這個刪除日志定位到具體的刪除代碼：
org.apache.pulsar.broker.service.persistent.PersistentTopic#checkReplication
圖片
原來是這里的 configuredClusters 值為空才導致的 topic 調用了 deleteForcefully()被刪除。
而這個值是從 topic 的 Policy 中獲取的。
復現問題
通過上圖中的堆棧跟蹤，懷疑是重啟 broker 導致的 topic unload ，同時 broker 又在構建 topic 導致了對 topicPolicy 的讀寫。
最終導致 topicPolicy 為空。
只要寫個單測可以復現這個問題就好辦了：
@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);
}
}同時還得查詢元數據有耗時才能復現：
圖片
只能手動 sleep 模擬這個耗時
具體也可以參考這個 issue https://github.com/apache/pulsar/issues/21653#issuecomment-1842962452
此時就會發現有 topic 被刪除了，而且是隨機刪除的，因為出現并發的幾率本身也是隨機的。
圖片
這里畫了一個流程圖就比較清晰了，在 broker 重啟的時候會有兩個線程同時topicPolicy 進行操作。
在 thread3 讀取 topicPolicy 進行判斷時，thread2 可能還沒有把數據準備好，所以就導致了 topic 被刪除。
修復
既然知道了問題原因就好修復了，我們只需要把 thread3 和 thread2 修改為串行執行就好了。
這也是處理并發最簡單高效的方法，就是直接避免并發；加鎖、隊列啥的雖然也可以解決，但代碼復雜度也高了很多，所以能不并發就盡量不要并發。
但要把這個修復推送到社區上游主分支最好是要加上單測，這樣即便是后續有其他的改動也能保證這個 bug 不會再次出現。
之后在社區大佬的幫助下完善了單測，最終合并了這個修復。
再次證明寫單測往往比代碼更復雜，也更花費時間。
PR：https://github.com/apache/pulsar/pull/21704
使用修復鏡像
因為社區合并代碼再發版的周期較長，而我們又急于修復該問題；不然都不敢重啟 broker，因為每重啟一次都可能會導致不知道哪個 topic 就被刪除了。
所以我們自己在本地構建了一個修復的鏡像，準備在線上進行替換。
此時坑又來了，我們滿懷信心的替換了一個鏡像再觀察日志發現居然還有刪除的日志??。
冷靜下來一分析，原來是當前替換進行的 broker 沒有問題了，但它處理的 topic 被轉移到了其他 broker 中，而其他的 broker 并沒有替換為我們最新的鏡像。
所以導致 topic 在其他 broker 中依然被刪除了。
除非我們停機，將所有的鏡像都替換之后再一起重啟。
但這樣的成本太高了，最好是可以平滑發布。
最終我們想到一個辦法，使用 arthas 去關閉了一個 broker 的一個選項，之后就不會執行出現 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 的評論區。https://github.com/apache/pulsar/issues/21653#issuecomment-1857548997如果不幸碰到了這個 bug，可以參考修復。
總結
刪除的這些 topic 的同時它的訂閱者也被刪除了，所以我們還需要修復訂閱者：
String topicName = "persistent://tenant/ns/topicName";
pulsarTopicService.createSubscription(topicName, "subName", MessageId.latest);之所以說這個 bug 非常致命，是因為這樣會導致 topic 的數據丟失，同時這些 topic 上的數據也會被刪除。
后續 https://github.com/apache/pulsar/pull/21704#issuecomment-1878315926社區也補充了一些場景。
圖片
其實場景 2 更容易出現復現，畢竟更容易出現并發；也就是我們碰到的場景
說來也奇怪，結合社區的 issue 和其他大佬的反饋，這個問題只有我們碰到了，估計也是這個問題的觸發條件也比較苛刻：
- 開啟 systemTopic/topicLevelPolicessystemTopicEnabled: "true" topicLevelPoliciesEnabled: "true"
- 設置足夠多的 topicPolicies
- 重啟 broker
重啟過程中從 zk 中獲取數據出現耗時
符合以上條件的集群就需要注意了。
其實這個問題在這個 PR 就已經引入了 https://github.com/apache/pulsar/pull/11021
所以已經存在蠻久了，后續我們也將檢測元數據作為升級流程之一了，確保升級后數據依然是完整的。
相關的 issue 和 PR：
https://github.com/apache/pulsar/issues/21653 https://github.com/apache/pulsar/pull/21704
