#作者：張桐瑞

主題日常管理

所謂的日常管理，無非就是主題的增刪改查。你可能會覺得，這有什么好討論的，官網上不都有命令嗎？這部分內容的確比較簡單，但它是我們討論后面內容的基礎。而且，在討論的過程中，我還會向你分享一些小技巧。另外，我們今天討論的管理手段都是借助于Kafka自帶的命令。事實上，在專欄后面，我們還會專門討論如何使用Java API的方式來運維Kafka集群。

我們先來學習一下如何使用命令創建Kafka主題。Kafka提供了自帶的kafka-topics腳本，用于幫助用戶創建主題。該腳本文件位于Kafka安裝目錄的bin子目錄下。如果你是在Windows上使用Kafka，那么該腳本位于bin路徑的windows子目錄下。一個典型的創建命令如下：
bin/kafka-topics.sh --bootstrap-server broker_host:port --create --topic my_topic_name --partitions 1 --replication-factor 1

create表明我們要創建主題，而partitions和replication factor分別設置了主題的分區數以及每個分區下的副本數。如果你之前使用過這個命令，你可能會感到奇怪：難道不是指定 –zookeeper參數嗎？為什么現在變成 –bootstrap-server了呢？我來給出答案：從Kafka 2.2版本開始，社區推薦用 –bootstrap-server參數替換 –zookeeper參數，并且顯式地將后者標記為“已過期”，因此，如果你已經在使用2.2版本了，那么創建主題請指定 –bootstrap-server參數。

社區推薦使用 –bootstrap-server而非 –zookeeper的原因主要有兩個。

1. 使用 –zookeeper會繞過Kafka的安全體系。這就是說，即使你為Kafka集群設置了安全認證，限制了主題的創建，如果你使用 –zookeeper的命令，依然能成功創建任意主題，不受認證體系的約束。這顯然是Kafka集群的運維人員不希望看到的。
2. 使用 –bootstrap-server與集群進行交互，越來越成為使用Kafka的標準姿勢。換句話說，以后會有越來越少的命令和API需要與ZooKeeper進行連接。這樣，我們只需要一套連接信息，就能與Kafka進行全方位的交互，不用像以前一樣，必須同時維護ZooKeeper和Broker的連接信息。
   創建好主題之后，Kafka允許我們使用相同的腳本查詢主題。你可以使用下面的命令，查詢所有主題的列表。
   bin/kafka-topics.sh --bootstrap-server broker_host:port --list
   如果要查詢單個主題的詳細數據，你可以使用下面的命令。

bin/kafka-topics.sh --bootstrap-server broker_host:port --describe --topic <topic_name>

如果describe命令不指定具體的主題名稱，那么Kafka默認會返回所有“可見”主題的詳細數據給你。

這里的“可見”，是指發起這個命令的用戶能夠看到的Kafka主題。這和前面說到主題創建時，使用 –zookeeper和 –bootstrap-server的區別是一樣的。如果指定了 –bootstrap-server，那么這條命令就會受到安全認證體系的約束，即對命令發起者進行權限驗證，然后返回它能看到的主題。否則，如果指定 –zookeeper參數，那么默認會返回集群中所有的主題詳細數據。基于這些原因，我建議你最好統一使用 –bootstrap-server連接參數。

說完了主題的“增”和“查”，我們說說如何“改”。Kafka中涉及到主題變更的地方有5處。

1. 修改主題分區。

其實就是增加分區，目前Kafka不允許減少某個主題的分區數。你可以使用kafka-topics腳本，結合 –alter參數來增加某個主題的分區數，命令如下：
bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic <topic_name> --partitions <新分區數>
這里要注意的是，你指定的分區數一定要比原有分區數大，否則Kafka會拋出InvalidPartitionsException異常。

2. 修改主題級別參數。

在主題創建之后，我們可以使用kafka-configs腳本修改對應的參數。
這個用法我們在專欄第8講中討論過，現在先來復習一下。假設我們要設置主題級別參數max.message.bytes，那么命令如下：
bin/kafka-configs.sh --zookeeper zookeeper_host:port --entity-type topics --entity-name <topic_name> --alter --add-config max.message.bytes=10485760
也許你會覺得奇怪，為什么這個腳本就要指定 –zookeeper，而不是 –bootstrap-server呢？其實，這個腳本也能指定 –bootstrap-server參數，只是它是用來設置動態參數的。在專欄后面，我會詳細介紹什么是動態參數，以及動態參數都有哪些。現在，你只需要了解設置常規的主題級別參數，還是使用 –zookeeper。

3. 變更副本數。

使用自帶的kafka-reassign-partitions腳本，幫助我們增加主題的副本數。這里先留個懸念，稍后我會拿Kafka內部主題__consumer_offsets來演示如何增加主題副本數。

4. 修改主題限速。

這里主要是指設置Leader副本和Follower副本使用的帶寬。有時候，我們想要讓某個主題的副本在執行副本同步機制時，不要消耗過多的帶寬。Kafka提供了這樣的功能。我來舉個例子。假設我有個主題，名為test，我想讓該主題各個分區的Leader副本和Follower副本在處理副本同步時，不得占用超過100MBps的帶寬。注意是大寫B，即每秒不超過100MB。那么，我們應該怎么設置呢？

要達到這個目的，我們必須先設置Broker端參數leader.replication.throttled.rate和follower.replication.throttled.rate，命令如下：

bin/kafka-configs.sh --zookeeper zookeeper_host:port --alter --add-config 'leader.replication.throttled.rate=104857600,follower.replication.throttled.rate=104857600' --entity-type brokers --entity-name 0

這條命令結尾處的 –entity-name就是Broker ID。倘若該主題的副本分別在0、1、2、3多個Broker上，那么你還要依次為Broker 1、2、3執行這條命令。

設置好這個參數之后，我們還需要為該主題設置要限速的副本。在這個例子中，我們想要為所有副本都設置限速，因此統一使用通配符*來表示，命令如下：

bin/kafka-configs.sh --zookeeper zookeeper_host:port --alter --add-config 'leader.replication.throttled.replicas=*,follower.replication.throttled.replicas=*' --entity-type topics --entity-name test

5.主題分區遷移。

同樣是使用kafka-reassign-partitions腳本，對主題各個分區的副本進行“手術”般的調整，比如把某些分區批量遷移到其他Broker上。這種變更比較復雜，我會在專欄后面專門和你分享如何做主題的分區遷移。
最后，我們來聊聊如何刪除主題。命令很簡單，我直接分享給你。
bin/kafka-topics.sh --bootstrap-server broker_host:port --delete --topic <topic_name>
刪除主題的命令并不復雜，關鍵是刪除操作是異步的，執行完這條命令不代表主題立即就被刪除了。它僅僅是被標記成“已刪除”狀態而已。Kafka會在后臺默默地開啟主題刪除操作。因此，通常情況下，你都需要耐心地等待一段時間。
特殊主題的管理與運維
說完了日常的主題管理操作，我們來聊聊Kafka內部主題consumer_offsets和transaction_state。前者你可能已經很熟悉了，后者是Kafka支持事務新引入的。如果在你的生產環境中，你看到很多帶有consumer_offsets和transaction_state前綴的子目錄，不用驚慌，這是正常的。這兩個內部主題默認都有50個分區，因此，分區子目錄會非常得多。
關于這兩個內部主題，我的建議是不要手動創建或修改它們，還是讓Kafka自動幫我們創建好了。不過這里有個比較隱晦的問題，那就是__consumer_offsets的副本數問題。

在Kafka 0.11之前，當Kafka自動創建該主題時，它會綜合考慮當前運行的Broker臺數和Broker端參數offsets.topic.replication.factor值，然后取兩者的較小值作為該主題的副本數，但這就違背了用戶設置offsets.topic.replication.factor的初衷。這正是很多用戶感到困擾的地方：我的集群中有100臺Broker，offsets.topic.replication.factor也設成了3，為什么我的__consumer_offsets主題只有1個副本？其實，這就是因為這個主題是在只有一臺Broker啟動時被創建的。

在0.11版本之后，社區修正了這個問題。也就是說，0.11之后，Kafka會嚴格遵守offsets.topic.replication.factor值。如果當前運行的Broker數量小于offsets.topic.replication.factor值，Kafka會創建主題失敗，并顯式拋出異常。

那么，如果該主題的副本值已經是1了，我們能否把它增加到3呢？當然可以。我們來看一下具體的方法。

第1步是創建一個json文件，顯式提供50個分區對應的副本數。注意，replicas中的3臺Broker排列順序不同，目的是將Leader副本均勻地分散在Broker上。該文件具體格式如下：

{"version":1, "partitions":[{"topic":"__consumer_offsets","partition":0,"replicas":[0,1,2]}, {"topic":"__consumer_offsets","partition":1,"replicas":[0,2,1]},{"topic":"__consumer_offsets","partition":2,"replicas":[1,0,2]},{"topic":"__consumer_offsets","partition":3,"replicas":[1,2,0]},...{"topic":"__consumer_offsets","partition":49,"replicas":[0,1,2]}
]}`

第2步是執行kafka-reassign-partitions腳本，命令如下：

bin/kafka-reassign-partitions.sh --zookeeper zookeeper_host:port --reassignment-json-file reassign.json --execute

除了修改內部主題，我們可能還想查看這些內部主題的消息內容。特別是對于__consumer_offsets而言，由于它保存了消費者組的位移數據，有時候直接查看該主題消息是很方便的事情。下面的命令可以幫助我們直接查看消費者組提交的位移數據。

bin/kafka-console-consumer.sh --bootstrap-server kafka_host:port --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

除了查看位移提交數據，我們還可以直接讀取該主題消息，查看消費者組的狀態信息。

bin/kafka-console-consumer.sh --bootstrap-server kafka_host:port --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$GroupMetadataMessageFormatter" --from-beginning

對于內部主題__transaction_state而言，方法是相同的。你只需要指定kafka.coordinator.transaction.TransactionLog$TransactionLogMessageFormatter即可。

6. 常見主題錯誤處理

最后，我們來說說與主題相關的常見錯誤，以及相應的處理方法。

常見錯誤1：主題刪除失敗。

當運行完上面的刪除命令后，很多人發現已刪除主題的分區數據依然“躺在”硬盤上，沒有被清除。這時該怎么辦呢？

實際上，造成主題刪除失敗的原因有很多，最常見的原因有兩個：副本所在的Broker宕機了；待刪除主題的部分分區依然在執行遷移過程。

如果是因為前者，通常你重啟對應的Broker之后，刪除操作就能自動恢復；如果是因為后者，那就麻煩了，很可能兩個操作會相互干擾。

不管什么原因，一旦你碰到主題無法刪除的問題，可以采用這樣的方法：
第1步，手動刪除ZooKeeper節點/admin/delete_topics下以待刪除主題為名的znode。

第2步，手動刪除該主題在磁盤上的分區目錄。

第3步，在ZooKeeper中執行rmr /controller，觸發Controller重選舉，刷新Controller緩存。
在執行最后一步時，你一定要謹慎，因為它可能造成大面積的分區Leader重選舉。事實上，僅僅執行前兩步也是可以的，只是Controller緩存中沒有清空待刪除主題罷了，也不影響使用。

常見錯誤2：__consumer_offsets占用太多的磁盤。

一旦你發現這個主題消耗了過多的磁盤空間，那么，你一定要顯式地用jstack命令查看一下kafka-log-cleaner-thread前綴的線程狀態。通常情況下，這都是因為該線程掛掉了，無法及時清理此內部主題。倘若真是這個原因導致的，那我們就只能重啟相應的Broker了。另外，請你注意保留出錯日志，因為這通常都是Bug導致的，最好提交到社區看一下。