「不要屈服，不要淡化，不要使它看來合邏輯，不要依據潮流而修改你的靈魂。相反的，狠狠的追隨你最強烈的喜好之事。」
“Don't bend; don't water it down; don't try to make it logical; don't edit your own soul according to the fashion. Rather, follow your most intense obsessions mercilessly.”
― Franz Kafka

昨天新增在單一個 broker 上新增了一個 topic，今天將要帶大家新增多個 broker，並且真的的實現 kafka 持久化。

Step 1: 設定並啟動三個 broker

1.1 調整 broker 設定檔

• 設定多個broker
  • 進入到 Kafka 資料夾中，將 server.properties 設定檔複製三份

$ cd /usr/local/etc/kafka
$ cp server.properties server-1.properties
$ cp server.properties server-2.properties
$ cp server.properties server-3.properties

• 依序修改設定檔
  • 我們要在同一台機器運作三個 broker，所以必須修改 port
  • 每個 broker 的 id 必須不同，這邊也須修改
  • log.dirs 就是讓訊息所謂持久化存放的位置，這邊也依照 broker 區分
  • 修改 server-1.properties

broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/usr/local/var/lib/kafka-logs/broker1

• 修改 server-2.properties

broker.id=1
listeners=PLAINTEXT://localhost:9093
log.dirs=/usr/local/var/lib/kafka-logs/broker2

• 修改 server-3.properties

broker.id=2
listeners=PLAINTEXT://localhost:9094
log.dirs=/usr/local/var/lib/kafka-logs/broker3

1.2 啟動三個 Broker

接下來，開啟三個終端機，分別啟動 broker1, broker2, broker3，這邊不用背景啟動，方便看 log 熟悉 broker server 實際上在運行什麼

$ cd  /usr/local/Cellar/kafka/2.8.0/libexec
$ kafka-server-start /usr/local/etc/kafka/server-1.properties
$ kafka-server-start /usr/local/etc/kafka/server-2.properties
$ kafka-server-start /usr/local/etc/kafka/server-3.properties

Step 2: 新增 Topic

2.1 這邊要用kafka-topics.sh去新增 topic

$ kafka-topics --create --zookeeper 127.0.0.1:2181 --replication-factor 2 --partitions 3 --topic topicWithThreeBroker

這邊跟昨天創建單一 broker 的 topic 有差別的參數是 replication-factor 跟 partitions

• --partitions 3 => 在 topic 下建立三個分區
• --replication-factor 2 => 意思是將每一個分區複製到其中兩個 broker 上

這樣創建的好處是什麼？
就算其中一個broker掛掉了，仍然可以從其他兩個broker獲取資料，但如果壞掉2個broker還是會掉資料，這樣的做法，只可以容忍1個broker壞掉

實際上的log目錄結構是這樣，可以清楚地看出這樣的配置是可以容忍N-1個 broker 壞掉：

* broker0
    * topic1-0
    * topic1-1
* broker1
    * topic1-1
    * topic1-2
* broker2
    * topic1-2
    * topic1-0

2.2 查看一下topic實際上是如何分配的

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 0	Replicas: 0,2	Isr: 0,2

參數介紹：
Partition: 0 => 代表 Topic topicWithThreeBroker 下編號為0的 Parition
Leader: 1 => 代表 partition 0 的 partition leader 在 broker1 上
=> 每個分區都有一個 leader，通常leader不會集中在同一個節點上
=> 這個 leader 會負責該 partition 內所有的讀寫，而其他的 broker 上的 replica(follower)，會去消費、同步該 leader 的資料
=> 如果這個 leader 掛掉了，那麼會從 Isr 中，選取第一個(先到先贏機制)成為該 partition 新的 leader
Replicas 1,0 => 代表該分區有副本在 broker1 和 broker0 上
Isr: 1, 0 => 代表我們當前可以在 broker1 和 broker0 上訪問該 partition0 的資料
=> 如果broker 0掛掉，Isr 會從 Isr: 1,0 變成 Isr: 1
=> Isr 的全名為in-sync replica，也就是已同步的副本

2.3 模擬其中一個 broker 壞掉後再恢復的情況

這邊將broker0 ctrl+c關掉，模擬少一個broker的情況，同樣去下指令

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 2	Replicas: 0,2	Isr: 2

這邊可以看到 Parition0 的 Isr 剩下 broker1、Partition2 的 Isr 剩下 broker2
而原本 Partition2 的 Leader 因為 Broker0 壞掉，會從 Isr 中選取，因此 Leader 變成 Broker2

把 broker0 重新啟動

$ kafka-server-start /usr/local/etc/kafka/server-1.properties

再下一次指令查看 Topic topicWithThreeBroker 的分配狀況

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 2	Replicas: 0,2	Isr: 2,0

現在可以看到 Partition0 和 Parition2 的 Isr 都有恢復了
Partition: 0 Isr: 2,0
Partition: 2 Isr: 1,0
也可以觀察到 Partition2 的 leader 沒有自動回歸到 broker0 上，還是在broker2上，這部分會在之後介紹如何重新分配 Partition Leader

Step 3: Broker 常用設定

Kafka cluster 會自動偵測是否有 broker 關閉或是失敗，然後幫那個 broker 上的 partition 重新選舉 leader，這個機制會在 broker 伺服器當掉、為了要維護而關掉、config檔案更改時生效。
Kafka 支援一個優雅的關閉機制去讓 server 停止運作，而非直接kill掉，這個機制會做兩件事：

1. 會將所有的log同步到磁碟中，避免重啟時還要去做log復原的動作，log復原需要時間，所以這個機制會加快 broker 重啟的速度。
2. 在將 broker 關閉之前，這個機制會移動在這個 broker 上是 leader 的 partition 到其他的 replica，這會加速 leader 轉移的時間、並將 partition 不可用的時間下降到幾毫秒(milliseconds); 除了hard kill之外，當 server 關閉或是失敗時觸發這個機制。

使用方式：
在檔案位置/usr/local/etc/kafka/server-1.properties加上設定controlled.shutdown.enable=true

需要注意的是 controlled shutdown 只會在broker上所有 partition 都有一個以上的 replicas 時才會成功。