人妻少妇精品视频一区二区三区,国产亚洲一区二区久久

在了解轉(zhuǎn)換過程之前還需了解 Pulsar Message 協(xié)議。協(xié)議中一部分信息專注于元數(shù)據(jù) ， message payload 字段中包含實(shí)際數(shù)據(jù)，每個(gè) message 中有多條消息，與 RecordBatch 類似。單條消息還有自己的元數(shù)據(jù) 。

從 Kafka 請求轉(zhuǎn)換為 Pulsar 消息要做協(xié)議轉(zhuǎn)換。當(dāng) entryFormat=kafka（取值為 kafka）時(shí)，主要會設(shè)置 publish time、num messages、properties（標(biāo)識 message 的 entryFormat 類型，解碼階段需要），最后 payload 部分就是將整個(gè) RecordBatch 通過 Persistent Topic 組件發(fā)送到 Broker 。這里 Pulsar 的客戶端無法識別解析 RecordBatch 。
如果要用 KoP 將 Kafka 集群數(shù)據(jù)遷移到 Pulsar，就需要用到 entryFormat=pulsar 。它會遍歷 Kafka 的 RecordBatch 和內(nèi)部的 Record 信息一一對應(yīng)設(shè)置能夠?qū)?yīng)的 Message Metadata 和 Single Message Metadata，從而轉(zhuǎn)換為 Pulsar 消息發(fā)布到 bookie 。
要解決兼容性問題就要專注于 EntryFormat，根據(jù)生產(chǎn)者和消費(fèi)者的版本情況進(jìn)行消息的轉(zhuǎn)換。轉(zhuǎn)換會出現(xiàn)性能損耗，此處注意消費(fèi)者版本較高時(shí)可以將轉(zhuǎn)換過程交給消費(fèi)者處理來節(jié)省性能。
新特性改進(jìn)介紹：元數(shù)據(jù)事件管理器
引入原因一：元數(shù)據(jù)不一致

上圖是一個(gè)兩節(jié)點(diǎn)的 KoP 集群，客戶端生產(chǎn)的 topic 的分區(qū) 0，位于 broker1 中。客戶端的引導(dǎo)地址是 broker1 和 broker2 ?，F(xiàn)在客戶端要發(fā)送元數(shù)據(jù)請求給 broker2，broker2 會響應(yīng) metadata response 。在 KoP 之前的處理邏輯中復(fù)用了 Topic lookup 機(jī)制， broker2 返回的 response 中不會包含自身的信息，只有分區(qū)所在的 broker1 的信息。然后客戶端會向 broker1 分區(qū)的 leader 節(jié)點(diǎn)發(fā)送生產(chǎn)者請求。
Broker1 掛掉后，分區(qū) 0 會容錯(cuò)到 Broker2 上。于是 broker2 成為分區(qū) 0 的 owner 。這時(shí)客戶端向 broker1 發(fā)送元數(shù)據(jù)請求失敗，又因?yàn)樽陨頉]有 broker2 的處理邏輯，所以元數(shù)據(jù)就無法路由到 broker2 上，出現(xiàn)元數(shù)據(jù)超時(shí)問題。

引入原因二：Group 殘留無效 topic 狀態(tài)

如上圖，通過 KoP 消費(fèi) topic 時(shí)，消費(fèi)的組元數(shù)據(jù)信息都會記到 coordinator 中，用 ./bin/kafka-consumer-groups.sh –bootstrap-server localhost:9092 –group my-group –describe可以看到描述的消費(fèi)信息。當(dāng) topic 下線并刪除時(shí)（./bin/kafka-topics.sh –bootstrap-server localhost:9092 —topic test —delete），再去描述組信息就會返回原數(shù)據(jù)超時(shí)異常。因?yàn)?admin 客戶端執(zhí)行刪除命令時(shí)，請求到達(dá) KoP Cluster，KoP broker 會通過 PulsarAdmin 刪除 topic 。Pulsar Cluster 處理刪除請求時(shí) ，會發(fā)送到所有分區(qū)的 owner broker 上，后者負(fù)責(zé)刪除 topic 信息并移除 topic 。
但因?yàn)?Group 元數(shù)據(jù)信息位于 coordinator 中，其 owner broker 和 topic owner broker 不在一起，所以刪除后者時(shí)無法清除前者，就會出現(xiàn)殘留。問題發(fā)生原因是 Group Coordinator 里面有 Group 元數(shù)據(jù)信息記錄了消費(fèi)分區(qū) ，客戶端在獲取分區(qū)時(shí) commit offset 會記錄 Lag 值，Kafka 當(dāng)前生產(chǎn)的消息位移。之后獲取 topic 信息，但是由于 topic 已經(jīng)刪除，因此會一直返回 onload partition 錯(cuò)誤。命令工具不斷重復(fù)嘗試獲取元數(shù)據(jù)直到 Request Timeout 超時(shí)并暴露超時(shí) 。