1.前言

為了盡量減少自然和人為災(zāi)難(如停電、災(zāi)難性軟件故障和網(wǎng)絡(luò)中斷)對(duì)業(yè)務(wù)的影響，以及隨著我行基于Kafka的實(shí)時(shí)業(yè)務(wù)不斷增長，Kafka的重要性日益增長，在我行逐步優(yōu)化跨IDC的Kafka連續(xù)性建設(shè)已經(jīng)成為我們目前亟待解決的問題。

本文就目前已有的災(zāi)備方案在元數(shù)據(jù)同步、數(shù)據(jù)復(fù)制、消費(fèi)位移同步、災(zāi)備模式等方面進(jìn)行調(diào)研對(duì)比。

2.現(xiàn)有災(zāi)備方案

3.各方案的主要設(shè)計(jì)點(diǎn)對(duì)比分析

3.1 元數(shù)據(jù)同步

元數(shù)據(jù)同步主要是指Topic、Partition、Configuration、ACL的同步，我們需要評(píng)估各方案在新增Topic，分區(qū)擴(kuò)容后、修改Configuration和ACL后能否自動(dòng)感知，以及評(píng)估方案中選擇復(fù)制的Topic是否靈活(比如是否支持白名單、黑名單機(jī)制，是否支持正則)，目標(biāo)集群中Topic名稱是否發(fā)生改變(決定是否支持雙向復(fù)制，是否會(huì)發(fā)生循環(huán)復(fù)制)。

MM1方案中，選擇復(fù)制的Topic只支持白名單機(jī)制(--whitelist或者--include參數(shù)指定)，且白名單支持正則寫法，但是當(dāng)源集群新增Topic后，目標(biāo)集群的auto.create.topics.enable設(shè)置為true時(shí)，才能自動(dòng)在目標(biāo)集群創(chuàng)建相同名稱的Topic(可以擴(kuò)展messagehandler改名)，否則必須重啟MirrorMaker才能發(fā)現(xiàn)新增的Topic，關(guān)于目標(biāo)集群上的Topic的分區(qū)數(shù)，MM1是按默認(rèn)值num.partitions進(jìn)行配置(其他方案均無該問題)，無法和源集群上保持一致，ACL也無法同步。

相比MM1，MM2彌補(bǔ)了上述不足，主要是依賴MirrorSourceConnector里的多個(gè)定時(shí)任務(wù)實(shí)現(xiàn)該功能，更新Topic/Partition、Configuration、ACL的間隔時(shí)長分別由三個(gè)參數(shù)指定，非常靈活。在MM2中，目前截至3.0.0的版本，支持兩種復(fù)制策略，默認(rèn)的DefaultReplicationPolicy中目標(biāo)集群中復(fù)制后Topic名稱發(fā)生變化，前面會(huì)加一個(gè)源集群的前綴，為了兼容MM1，3.0.0中新增的IdentityReplicationPolicy中目標(biāo)集群中復(fù)制后Topic名稱不會(huì)發(fā)生變化。

Confluent Replicator，根據(jù)官網(wǎng)描述，也同樣具備上述功能，原理和MM2類似，只是檢測更新只由一個(gè)參數(shù)確定。Replicator可以定義復(fù)制后Topic的名稱，由參數(shù)topic.rename.format指定，默認(rèn)值是保持Topic名稱不變。

基于Follower的同步機(jī)制的方案，由于網(wǎng)上資料不足，具體實(shí)現(xiàn)無法得知，但是原理估計(jì)和MM2類似，復(fù)制后在目標(biāo)集群中Topic名稱保持不變。

uReplicator的實(shí)現(xiàn)略有不同，復(fù)制哪些Topic，由參數(shù)enableAutoWhitelist和patternToExcludeTopics一起決定，當(dāng)enableAutoWhitelist設(shè)置為true時(shí)，若源集群和目標(biāo)集群中存在相同Topic，那么不需要其他設(shè)置即可實(shí)現(xiàn)數(shù)據(jù)復(fù)制，若設(shè)置為false，需要將復(fù)制的Topic名稱等信息提交給uReplicator Controller，由該Controller來控制分區(qū)的分配，另外黑名單參數(shù)patternToExcludeTopics控制哪些Topic不用復(fù)制；分區(qū)擴(kuò)容是否自動(dòng)感知，是由參數(shù)enableAutoTopicExpansion控制的；關(guān)于Configuration和ACL無法實(shí)現(xiàn)同步。

brooklin選擇復(fù)制的Topic只支持白名單機(jī)制，可支持正則，新增Topic和分區(qū)擴(kuò)容后可自動(dòng)感知，檢測更新由參數(shù)partitionFetchIntervalMs確定，復(fù)制后Topic名稱前可加前綴，由參數(shù)DESTINATION_TOPIC_PFEFIX確定。

總結(jié)如下：

3.2 數(shù)據(jù)復(fù)制

數(shù)據(jù)復(fù)制是災(zāi)備方案的最核心點(diǎn)之一，我們需要評(píng)估各方案中復(fù)制后消息offset能否對(duì)齊，復(fù)制期間數(shù)據(jù)的一致性能否保證即是否會(huì)丟失數(shù)據(jù)或者會(huì)出現(xiàn)重復(fù)數(shù)據(jù)。首先說明一下，由于復(fù)制會(huì)有延遲，因此所有這些災(zāi)備方案里RPO都不等于0。

基于Follower的同步機(jī)制的方案可以保持offset對(duì)齊，由于副本同步存在延遲，當(dāng)主機(jī)房異常時(shí)，備機(jī)房上仍有丟失部分?jǐn)?shù)據(jù)的可能性，offset可保持一致，不會(huì)出現(xiàn)重復(fù)數(shù)據(jù)的可能性。其他方案均不能保證offset對(duì)齊(除非是復(fù)制時(shí)源Topic的offset從0開始)，關(guān)于每個(gè)方案中消費(fèi)者從源集群消費(fèi)，再寫入到目標(biāo)集群的邏輯，我們一一詳細(xì)解釋下：

先從MM1開始，這是他的設(shè)計(jì)架構(gòu)：

在KIP-3 MirrorMaker Enhancement里，設(shè)計(jì)了上述架構(gòu)，從以下幾處保證不丟數(shù)：

1.關(guān)掉消費(fèi)者的自動(dòng)提交位移，提交位移之前會(huì)調(diào)用producer.flush()刷出緩存里數(shù)據(jù)

2.在producer端，通過設(shè)置這幾個(gè)參數(shù)max.in.flight.requests.per.connection=1(多個(gè)consumer共享一個(gè)producer，這個(gè)producer每次只給broker發(fā)一個(gè)request),retries=Int.MaxValue(返回是可重試異常，無限次重試直到緩沖區(qū)滿)，ack=-1(發(fā)給所有副本)

3.設(shè)置abortOnSendFail，當(dāng)producer端收到不可重試異常后(比如消息過大之類的異常)，停止MirrorMaker進(jìn)程，否則會(huì)丟失發(fā)送失敗的部分?jǐn)?shù)據(jù)

另外為了避免在consumer發(fā)生rebalance的是時(shí)候出現(xiàn)重復(fù)數(shù)據(jù)(rebalance時(shí)候有些數(shù)據(jù)位移沒提交)，定義了一個(gè)新的consumerRebalance監(jiān)聽器，在發(fā)生partitionRevoke的時(shí)候，先刷出producer緩存里數(shù)據(jù)，再提交位移。

從上面設(shè)計(jì)來看，MM1是不丟數(shù)，但是還是存在數(shù)據(jù)重復(fù)的可能性，這是Kafka的非冪等Producer決定的，另外MM1的設(shè)計(jì)還有很多缺陷，比如只有一個(gè)Producer，發(fā)送效率低，另外這個(gè)Producer是輪詢發(fā)送，消息發(fā)送到目的Topic上的分區(qū)和源Topic的分區(qū)不一定一致，由于是輪詢，這個(gè)Producer和集群里每個(gè)broker會(huì)建立連接。對(duì)比uReplicator，同樣也是在flush之后再提交位移去避免丟數(shù)，在MM1的缺陷都得到了改進(jìn)，每個(gè)WorkerInstance里有多個(gè)FetcherThread和多個(gè)ProducerThread，從源集群fetch數(shù)據(jù)后會(huì)放到一個(gè)隊(duì)列里，ProducerThread從隊(duì)列里取走數(shù)據(jù)并發(fā)到目標(biāo)集群的Topic，每條消息發(fā)送到目的Topic上分區(qū)和源分區(qū)保持一致，可以保持語義上一致。

在brooklin中，每個(gè)Brooklin Instance中可以起多個(gè)Consumer和Producer，也可保持語義上一致，比uReplicator更有優(yōu)勢(shì)的一處就是提供了flushless的生產(chǎn)者(也可提供flush的Producer)，哪些消息發(fā)送成功，才會(huì)提交這些位移，因?yàn)檎{(diào)用Producer.flush()可以將緩沖區(qū)的數(shù)據(jù)強(qiáng)制發(fā)送，但是代價(jià)較高，在清空緩沖前會(huì)堵塞發(fā)送線程。

consumer.poll()->producer.send(records)->producer.flush()->consumer.commit()
優(yōu)化為：
consumer.poll()->producer.send(records)->consumer.commit(offsets)

在MirrorMaker2中，采用Kafka Connect框架進(jìn)行復(fù)制數(shù)據(jù)，從源端消費(fèi)數(shù)據(jù)后，存到一個(gè)類型為IdentityHashMap的內(nèi)存結(jié)構(gòu)outstandingMessages中，Producer發(fā)送到目的端成功后，會(huì)從該內(nèi)存結(jié)構(gòu)中刪除該消息，另外會(huì)定時(shí)將從源端消費(fèi)的進(jìn)度保存到Kafka Topic中。這種實(shí)現(xiàn)機(jī)制不會(huì)丟失數(shù)據(jù)，但是Producer發(fā)送成功后，未將進(jìn)度持久化前進(jìn)程異常掛掉，那么會(huì)產(chǎn)生重復(fù)消息。目前在KIP-656: MirrorMaker2 Exactly-once Semantics提出了一種可實(shí)現(xiàn)Exactly Only Once的方案，思路是將提交消費(fèi)位移和發(fā)送消息放在一個(gè)事務(wù)里，但是相關(guān)Patch KAFKA-10339仍然沒被合進(jìn)主分支，最后更新停留在20年8月份。

根據(jù)Confluent Replicator官網(wǎng)描述，復(fù)制不會(huì)丟數(shù)，但是可能會(huì)重復(fù)，因此和上述MM2、uReplicator、brooklin一樣，提供的都是At least Once Delivery消息傳遞語義。

3.3 消費(fèi)位移同步

災(zāi)備方案中除數(shù)據(jù)復(fù)制，消費(fèi)位移的同步也非常關(guān)鍵，災(zāi)備切換后消費(fèi)者是否能在新的集群中恢復(fù)消費(fèi)，取決于consumer offset是否能同步。

在MM1設(shè)計(jì)中，若要同步消費(fèi)位移，只能將__consumer_offsets作為一個(gè)普通的Topic進(jìn)行同步，但是由于源集群和目標(biāo)集群的offset可能存在不對(duì)齊的情況，因此無法進(jìn)行offset轉(zhuǎn)換。

在MM2設(shè)計(jì)中，解決了上述MM1問題，設(shè)計(jì)思路是會(huì)定期在目標(biāo)集群的checkpoint Topic中記錄消費(fèi)位移，包括源端和目標(biāo)端的已提交位移，消息包括如下字段：

• consumer group id (String) 消費(fèi)組
• topic (String) – includes source cluster prefix topic名稱
• partition (int)  分區(qū)名稱
• upstream offset (int): latest committed offset in source cluster  源集群的消費(fèi)位移
• downstream offset (int): latest committed offset translated to target cluster 目標(biāo)集群的消費(fèi)位移
• metadata (String) partition元數(shù)據(jù)
• timestamp

另外，還設(shè)計(jì)了一個(gè)offset sync Topic用于記錄源端和目的端offset的映射。

同時(shí)，MM2還提供了MirrorClient接口做位移轉(zhuǎn)換：

// Find the local offsets corresponding to the latest checkpoint from a specific upstream consumer group.
Map<TopicPartition, OffsetAndMetadata> remoteConsumerOffsets(String consumerGroupId,
      ``String remoteClusterAlias, Duration timeout) ...

在uReplicator中，另外設(shè)計(jì)了一個(gè)offset Sync的服務(wù)，跟MM2類似(可能是MM2參考了uReplicator的設(shè)計(jì))，這個(gè)服務(wù)可以實(shí)時(shí)收集不同集群offset 的映射關(guān)系，計(jì)算出從一個(gè)DC切換到另一個(gè)DC后需要從哪個(gè) offset 進(jìn)行讀取。

在brooklin中，沒有類似uReplicator里的offset Sync服務(wù)，需要自己實(shí)現(xiàn)。

在Confluent Replicator中，用另外一種思路解決該問題，不同DC的時(shí)間是一致的，在Kafka的消息里包含時(shí)間戳，5.0 版引入了一項(xiàng)新功能，該功能使用時(shí)間戳自動(dòng)轉(zhuǎn)換偏移量，以便消費(fèi)者可以故障轉(zhuǎn)移到不同的數(shù)據(jù)中心并開始在目標(biāo)集群中消費(fèi)他們?cè)谠醇褐兄袛嗟臄?shù)據(jù)。要使用此功能，需要在Consumer中設(shè)置Consumer Timestamps Interceptor 的攔截器，該攔截器保留消費(fèi)消息的元數(shù)據(jù)，包括：

? Consumer group ID

? Topic name

? Partition

? Committed offset

? Timestamp

此消費(fèi)者時(shí)間戳信息保存在位于源集群中名為 __consumer_timestamps 的 Kafka  Topic中。然后Replicator通過以下步驟進(jìn)行offset轉(zhuǎn)換：

1. 從源集群中的 consumer_timestamps 主題中讀取消費(fèi)者偏移量和時(shí)間戳信息，以獲取消費(fèi)者組的進(jìn)度
2. 將源數(shù)據(jù)中心中的已提交偏移量轉(zhuǎn)換為目標(biāo)數(shù)據(jù)中心中的相應(yīng)偏移量
3. 將轉(zhuǎn)換后的偏移量寫入目標(biāo)集群中的 __consumer_offsets 主題

那么消費(fèi)者切換到目標(biāo)中心的集群后，可繼續(xù)進(jìn)行消費(fèi)。

基于Follower的同步機(jī)制方案，Topic完全一致，只要將__consumer_offsets也同步，那么消費(fèi)者故障轉(zhuǎn)移后仍可繼續(xù)消費(fèi)。

在消費(fèi)位移同步方面，各方案總結(jié)如下：

3.4 是否支持雙活

為了提升資源利用率，災(zāi)備模式的選取也是一個(gè)重要考量點(diǎn)。

MM1是不支持雙活模式的，兩個(gè)集群無法配置為相互復(fù)制（“Active/Active”），主要是因?yàn)槿绻趦蓚€(gè)集群中若存在相同名稱的Topic，無法解決Topic循環(huán)復(fù)制的問題。

MM1這個(gè)可能循環(huán)復(fù)制的問題在MM2中解決，解決思路是復(fù)制后的Topic與原Topic名稱不一致，會(huì)加上源集群的名稱作為前綴，例如如下示例中，A集群中的topic1在復(fù)制到B集群后，名稱變更為A.topic1。

但是MM2默認(rèn)的DefaultReplicationPolicy是復(fù)制后Topic名稱改變，對(duì)客戶端來說會(huì)增加切換代價(jià)，可以考慮改成IdentityReplicationPolicy，這種復(fù)制策略只能支持單向復(fù)制，主集群提供業(yè)務(wù)服務(wù)，即Active/Standy模式。

在Confluent Replicator 5.0.1中，為了避免循環(huán)復(fù)制，利用了KIP-82 Add Record Headers的特性，在消息的header里加入了消息來源，如果目標(biāo)集群的集群 ID 與header里的源集群 ID 匹配，并且目標(biāo)Topic名稱與header的Topic名稱匹配，則 Replicator 不會(huì)將消息復(fù)制到目標(biāo)集群。如下圖所示：

DC-1的m1復(fù)制后DC-2，消息的header里加入了標(biāo)記，這條消息是從DC-1復(fù)制過來的，那么Replicator不會(huì)把DC-2的m1再復(fù)制到DC-1，同理，DC-1的m2也不會(huì)復(fù)制到DC-2。因此Confluent Replicator是可以支持Active/Active模式的。

在uReplicator中，通過數(shù)據(jù)的冗余提供Region級(jí)別的故障轉(zhuǎn)移，在這種設(shè)計(jì)中，每個(gè)區(qū)域除部署一套本地Kafka集群，還會(huì)部署一套聚合集群，這套聚合集群里存儲(chǔ)了所有區(qū)域的數(shù)據(jù)。

當(dāng)區(qū)域集群A和B中存在相同Topic，那么匯聚后，在區(qū)域A和B中的消息offset可能不一致，uReplicator設(shè)計(jì)了一個(gè)offset管理服務(wù)，會(huì)記錄這個(gè)對(duì)應(yīng)關(guān)系，示例如下：

這種設(shè)計(jì)中，可以支持消費(fèi)者的Active/Active和Active/Standy模式，前者是每個(gè)區(qū)域起一個(gè)消費(fèi)者消費(fèi)聚合集群的的數(shù)據(jù)，只有一個(gè)區(qū)域是主區(qū)域，只有主區(qū)域的數(shù)據(jù)可以更新數(shù)據(jù)到后端數(shù)據(jù)庫中，當(dāng)主區(qū)域故障后，指定新的主區(qū)域，新的主區(qū)域繼續(xù)消費(fèi)計(jì)算，在Active/Standy模式中，所有區(qū)域中只有一個(gè)消費(fèi)者，該區(qū)域故障后，在其他區(qū)域啟動(dòng)一個(gè)消費(fèi)者，根據(jù)offset管理服務(wù)里記錄的offset對(duì)應(yīng)關(guān)系，從每個(gè)區(qū)域的區(qū)域集群中找到所有最新的checkpoints，然后根據(jù)該checkpoints在Standy區(qū)域的聚合集群查找最小offset，從Standy區(qū)域的該offset開始消費(fèi)。

在brooklin中，也可以通過類似uReplicator的設(shè)計(jì)利用數(shù)據(jù)的冗余實(shí)現(xiàn)Active/Active災(zāi)備模式。

在字節(jié)介紹的災(zāi)備方案中，Producer只能往主集群寫(主備集群中的信息是存儲(chǔ)在配置中心里的，客戶端需要先從配置中心查詢)，Producer可以在雙中心部署，但是通過配置中心路由到主集群，Consumer也可在雙中心部署，若采用Active/Standy模式，各自消費(fèi)本地機(jī)房的數(shù)據(jù)，但是只有主集群里消費(fèi)者的消費(fèi)位移可以生效，在采用Active/Active模式下，消費(fèi)者只能從主集群進(jìn)行消費(fèi)，這兩種模式下，都是將雙中心所有消費(fèi)者的消費(fèi)位移采用一個(gè)存儲(chǔ)統(tǒng)一存儲(chǔ)。

在災(zāi)備模式方面，各方案總結(jié)如下：

4.各方案的主要設(shè)計(jì)點(diǎn)總結(jié)

總結(jié)來說，這些方案里歸結(jié)為三類：

1.Kafka社區(qū)的設(shè)計(jì)路線方案，從源集群消費(fèi)，再寫入到目標(biāo)集群，包含MM1，MM2，uReplicator，brooklin這幾種方案，MM2是參考了uReplicator的設(shè)計(jì)，實(shí)現(xiàn)方案和brooklin類似，那么在這四種方案中，MM2可以作為優(yōu)先考慮方案。

2.Confluent Replicator的商業(yè)收費(fèi)方案，也是利用Kafka Connect框架進(jìn)行消費(fèi)寫入，在避免Topic循環(huán)復(fù)制和消費(fèi)位移轉(zhuǎn)換方面做得非常出色，客戶端切換的代價(jià)很低。

3.以字節(jié)、滴滴為代表的基于Follower同步機(jī)制的方案，這種方案里復(fù)制后的Topic是源Topic的鏡像，客戶端不需要做offset轉(zhuǎn)換，需要改造Kafka代碼，考慮到后續(xù)和原生Kafka代碼的版本融合，技術(shù)要求較高。

目前來說，沒有一個(gè)完美的解決方案，各公司可根據(jù)自身實(shí)際需求制定。