Kafka通訊協(xié)議指南

Kafka中英文術(shù)語比照為避開歧義，大局部的英文術(shù)語找不到適宜中文對應(yīng)時都保持英文原文，Kafka中一些根本術(shù)語也使用英文，其中一局部通過括號參與英文原文；另外，文中可能使用到的中英文術(shù)語包括但不限于：MetadataoffsetComsumer

ComsumerGroupTopic 主題API 接口Coordinator 協(xié)調(diào)器簡介此文檔涵蓋了Kafka0.8一個包含的可懇求的協(xié)議及其二進(jìn)制格式以及如何正確使用他們來實(shí)現(xiàn)一個客戶端的通訊協(xié)議文檔。本文假設(shè)您已經(jīng)了解了Kafka根本的設(shè)計(jì)以及術(shù)語。0.7和更早的版本所使用的協(xié)議與此類似，但我們〔期望〕通過一次性地?cái)財(cái)嗉嫒菪?，以便清理原有設(shè)計(jì)上的沉疴，并且泛化一些概念。假設(shè)遇到無法理解的狀況，請參照英文原文概述卡夫卡協(xié)議是相當(dāng)簡潔的，只有六個核心的客戶端懇求的API：元數(shù)據(jù)〔Metadata〕–描述可用的brokers，包括他們的主機(jī)broker發(fā)送〔Send〕–發(fā)送消息到broker；獵取〔Fetch〕–從brokertopic偏移量〔Offsets〕–獵取給定topic的分區(qū)的可用偏移量信息；偏移量提交〔OffsetCommit〕–提交消費(fèi)者組〔ComsumerGroup〕的一組偏移量；偏移量獵取〔OffsetFetch〕–獵取一個消費(fèi)者組的一組偏移量；上述的API都將在下面具體說明。此外，從0.9版本開頭，Kafka支持為消費(fèi)者和Kafka連接進(jìn)展分組治理。客戶端API包括五個懇求：分組協(xié)調(diào)者〔GroupCoordinator〕–用來定位一個分組當(dāng)前的協(xié)調(diào)者。參與分組〔JoinGroup〕–成為某一個分組的一個成員，當(dāng)分組不存在〔沒有一個成員時〕創(chuàng)立分組。同步分組〔SyncGroup〕–同步分組中全部成員的狀態(tài)〔例如分發(fā)分區(qū)安排信息(PartitionAssignments)到各個組員〕。心跳〔Heartbeat〕–保持組內(nèi)成員的活潑狀態(tài)。離開分組〔LeaveGroup〕–直接離開一個組。最終，有幾個治理API，可用于監(jiān)控/治理的卡夫卡集群〔KIP-4完成時，這個列表將增長〕：描述消費(fèi)者組〔DescribeGroups〕–用于檢查一組群體的當(dāng)前狀態(tài)〔如：查看消費(fèi)者分區(qū)安排〕。1.列出組〔ListGroups〕–列出某一個broker組開頭網(wǎng)絡(luò)Kafka使用基于TCP的二進(jìn)制協(xié)議。該協(xié)議定義了全部API的懇求及響應(yīng)消息。全部消息都是有長度限制的，并且由后面描述的根本類型組成?？蛻舳藛拥膕ocket連接，并且寫入懇求的消息序列和讀回相應(yīng)的響應(yīng)消息。連接和斷開時均不需要握手消息。假設(shè)保持你保持長連接，那么TCP協(xié)議本身將會節(jié)約很多TCP握手時間，但假設(shè)真的重建立連接，那么代價(jià)也相當(dāng)小?？蛻艨赡苄枰S持到多個broker的連接，由于數(shù)據(jù)是被分區(qū)的，而客戶端需要和存儲這些分區(qū)的broker效勞器進(jìn)展通訊。固然，一般而言，不需要為單個效勞端和單個客戶端間維護(hù)多個連接〔即連接池技術(shù)〕。效勞器的保證單一的TCP連接中，懇求將被挨次處理，響應(yīng)也將按該挨次返回。為保證broker的處理懇求的挨次，單個連接同時也只會處理一個懇求指令。請留意，客戶端可以〔也應(yīng)當(dāng)〕使用非堵塞IO實(shí)現(xiàn)懇求流水線，從而實(shí)現(xiàn)更高的吞吐量。也就是說，客戶可以在等待上次懇求應(yīng)答的同時發(fā)送下個懇求，由于待完成的懇求將會在底層操作系統(tǒng)套接字緩沖區(qū)進(jìn)展緩沖。除非特別說明，全部的懇求是由客戶端啟動，并從效勞器獵取到相應(yīng)的響應(yīng)消息。效勞器能夠配置懇求大小的最大限制，超過這個限制將導(dǎo)致socket分區(qū)和引導(dǎo)〔Partitioningandbootstrapping〕Kafka是一個分區(qū)系統(tǒng)，所以不是全部的效勞器都具有完整的數(shù)據(jù)集。主題(Topic)被分為P〔預(yù)先定義的分區(qū)數(shù)量〕個分區(qū)，每個分區(qū)被復(fù)制N〔復(fù)制因子〕份，TopicPartition依據(jù)挨次在“提交日志”0，1，…，P。全部具有這種特性的系統(tǒng)都有一個如何制定某個特定數(shù)據(jù)應(yīng)當(dāng)被安排給哪個特定的分區(qū)的問題。Kafka中它由客戶端直接把握安排策略，broker則沒有特別的語義來打算消息公布到哪個分區(qū)。相反，生產(chǎn)者直接將消息發(fā)送到一個特定的分區(qū)，提取消息時，消費(fèi)者也直接從某個特定的分區(qū)獵取。假設(shè)兩個生產(chǎn)者要使用一樣的分區(qū)方案，那么他Key這些公布或獵取數(shù)據(jù)的懇求必需發(fā)送到指定分區(qū)中作為leader的broker。此條件同時也會由broker保證，發(fā)送到不正確的broker的懇求將會返回NotLeaderForPartition錯誤代碼〔后文所描述的〕。那么客戶端如何找出哪些主題存在，他們有什么分區(qū)，以及這些分區(qū)被哪些broker存取，以便它可以直接將懇求發(fā)送到所在的主機(jī)？這個信息是動態(tài)的，因此你不能只是供給每個客戶端一些靜態(tài)映射文件。全部的Kafkabroker都可以答復(fù)這個描述集群的當(dāng)前狀態(tài)的數(shù)據(jù)懇求：有哪些主題，這些主題都有多少分區(qū)，哪個broker是這些分區(qū)Leaderbroker換句話說，客戶端只需要找到一個broker，broker將會告知客戶端全部其他存在的broker，以及這些broker上面的全局部區(qū)。這個broker本身也可能會掉線，因此客戶端實(shí)現(xiàn)的最正確做法是保存兩個或三個broker地址，從而來引導(dǎo)列表。用戶可以選擇使用負(fù)載均衡器或Kafka客戶并不需要輪詢地查看集群是否已經(jīng)轉(zhuǎn)變；它可以等到它接收到所用的元數(shù)據(jù)是過時的錯誤信息時一次性更元數(shù)據(jù)。這中錯誤有兩種形式：〔1〕一個套接字錯誤指示客戶端不能與特定的broker進(jìn)展通信，〔2〕懇求響應(yīng)說明該 broker不再是其懇求數(shù)據(jù)分區(qū)的Leader的錯誤。輪詢“起始”Kafka的URL列表，直到我們找到一個我們可以broker。獵取集群元數(shù)據(jù)。處理獵取數(shù)據(jù)或者存儲消息懇求，依據(jù)這些懇求所發(fā)送的主題broker。假設(shè)我們得到一個適當(dāng)?shù)腻e誤(顯示元數(shù)據(jù)已經(jīng)過時時)，刷元數(shù)據(jù)，然后再試一次。分區(qū)策略〔PartitioningStrategies〕上面提到消息的分區(qū)安排是由生產(chǎn)者客戶端把握，那么，為什么要把這個功能被暴露給最終用戶？Kafka它平衡了broker的數(shù)據(jù)和懇求負(fù)載它允很多個消費(fèi)者之間處理分發(fā)消息的同時，能夠維護(hù)本地狀態(tài)，并且在分區(qū)中維持消息的挨次。我們稱這種語義的分區(qū)〔semanticpartitioning〕。對于給定的使用場景下，你可能只關(guān)心其中的一個或兩個。為了實(shí)現(xiàn)簡潔的負(fù)載均衡，一個簡潔的策略是客戶端公布消息是對全部broker進(jìn)展輪詢懇求(roundrobinrequests)。另一種選擇，在那些生產(chǎn)者比消費(fèi)者多的場景下，給每個客戶機(jī)隨機(jī)選擇并公布消息到該分區(qū)。后一種的策略能夠使用少得多的TCP語義分區(qū)是指使用關(guān)鍵字(key)來打算消息安排的分區(qū)。例如，假設(shè)你正在處理一個點(diǎn)擊消息流時，可能需要通過用戶ID來劃分流，使得特定用戶的全部數(shù)據(jù)會被單個消費(fèi)者消費(fèi)。要做到這一點(diǎn)，客戶端可以實(shí)行與消息相關(guān)聯(lián)的關(guān)鍵字，并使用關(guān)鍵字的某個Hash值來選擇的傳送的分區(qū)。批處理〔Batching〕我們的API鼓舞將小的懇求批量處理以提高效率。我們覺察這能格外顯著地提升性能。我們兩個用來發(fā)送消息和獵取消息的API，總是以一連串的消息工作，而不是單一的消息，從而鼓舞批處理操作。聰明的客戶端可以利用這一點(diǎn)，并支持“異步”操作模式，以此進(jìn)展批處理哪些單獨(dú)發(fā)送的消息，并把它們以較大的塊進(jìn)展發(fā)送。我們再進(jìn)一步允許跨多個主題和分區(qū)的批處理，所以生產(chǎn)懇求可能包含追加到很多分區(qū)的數(shù)據(jù)，一個讀取懇求可以一次性從多個分區(qū)提取數(shù)據(jù)的。固然，假設(shè)他們寵愛，客戶端實(shí)現(xiàn)者可以選擇無視這一點(diǎn)，全部消息一次都發(fā)送一個。版本和兼容性〔VersioningandCompatibility〕該協(xié)議的目的要到達(dá)在向后兼容的根底上漸進(jìn)演化。我們的版本是基于每個API根底之上，每個版本包括一個懇求和響應(yīng)對。每個懇求APIKey，里面包含了被調(diào)用的API標(biāo)識，以及表示這些懇求和響應(yīng)格式的版本號。這樣做的目的是允許客戶端執(zhí)行相應(yīng)特定版本的懇求。目標(biāo)主要是為了在不允許停機(jī)的環(huán)境下進(jìn)展更，這種環(huán)境下，客戶端和效勞API。效勞器將拒絕它不支持的版本的懇求，并始終返回它期望收到的能夠完成懇求響應(yīng)的版本的協(xié)議格式。預(yù)期的升級路徑方式是，功能將首先部署到效勞器〔老客戶端無法完全利用他們的功能〕，然后隨著的客戶端的部署，這些功能將逐步被利用。目前，全部版本基線為0，當(dāng)我們演進(jìn)這些API時，我們將分別顯示每個版本的格式。通訊協(xié)議〔TheProtocol〕協(xié)議根本數(shù)據(jù)類型〔ProtocolPrimitiveTypes〕Theprotocolisbuiltoutofthefollowingprimitivetypes.該協(xié)議是建立在以下根本類型之上。定長根本類型〔FixedWidthPrimitives〕int8,int16,int32,int64–不同精度(以bit數(shù)區(qū)分)的帶符號整數(shù)，以大端〔BigEndiam〕方式存儲.變長根本類型〔VariableLengthPrimitives〕bytes,string–這些類型由一個表示長度的帶符號整數(shù)N以及后續(xù)N字節(jié)的內(nèi)容組成。長度假設(shè)為-1表示空〔null〕.string使用int16bytesint32.數(shù)組〔Arrays〕這個類型用來處理重復(fù)的構(gòu)造體數(shù)據(jù)。他們總是由一個代表元素個數(shù)int32整數(shù)N，以及后續(xù)的N個重復(fù)構(gòu)造體組成，這些構(gòu)造體自身是有其他的根本數(shù)據(jù)類型組成。我們后面會用BNF語法呈現(xiàn)一個foo[foo]懇求格式語法要點(diǎn)〔Notesonreadingtherequestformatgrammars〕后面的BNF精準(zhǔn)地以上下文無關(guān)的語法呈現(xiàn)了懇求和響應(yīng)的二進(jìn)制格式。每個API都會一起給出懇求和響應(yīng)，以及全部的子定義〔sub-definitions〕。BNF使用沒有經(jīng)過縮寫的便于閱讀的名稱〔比方我使用一個符號化了得名稱來定義了一個生產(chǎn)者錯誤碼，即便它只是int16整數(shù)〕。一般在BNF中，一個序列表示一個連接，所以下面給出的MetadataRequest將是一個含有VersionId，然后clientId，然后TopicNames的陣列〔每一個都有其自身的定義〕。自定義類型一般使用駝峰法拼寫，根本類型使用全小寫方式乒協(xié)。當(dāng)存在多中可能的自定義類型時，使用’|’符號分割，并且用括號表示分組。頂級定義不縮進(jìn)，后續(xù)的子局部會被縮進(jìn)。一般的懇求和響應(yīng)格式〔CommonRequestandResponseStructure〕全部懇求和響應(yīng)都從以下語法起源，其余的會在本文剩下局部中進(jìn)展增量描述：1.2.RequestOrResponseResponseMessage)3.4.5.Size=>int326.域〔FIELD〕

=> Size (RequestMessage |描述MessageSizeMessageSize域給出了 后續(xù)懇求或響應(yīng)消息的字節(jié)(bytes)長度?？蛻舳丝梢韵茸x取4字節(jié)的長度N，然后讀取并解析后續(xù)的N字節(jié)懇求內(nèi)容。懇求〔Requests〕全部懇求都具有以下格式：1.2.RequestMessage=> ApiKeyClientIdRequestMessage3.4.5.ApiKey=>int166.7.8.ApiVersion=>int169.10.11. CorrelationId=>int3212.

ApiVersionCorrelationId13.14.15.

ClientId=>string16.17. RequestMessage

=> MetadataRequest |ProduceRequest |

FetchRequest

| OffsetRequest |OffsetCommitRequest|OffsetFetchRequest18.ApiKeyApiVersion

描述id〔即它表示是一個元數(shù)據(jù)懇求？生產(chǎn)懇求？獵取懇求等〕.API號，該版本號允許效勞器依據(jù)版本號正確地解釋懇求內(nèi)容。響API0?？蛻舳?。用于匹配客戶機(jī)和效勞器之間的懇求和響應(yīng)。ClientId

場景。例如，你可能不僅想要監(jiān)視每秒的總體懇求，還要依據(jù)客戶端應(yīng)用程序進(jìn)展監(jiān)視，那它就可以被用上〔其中每一個都將駐留在多個效勞器上〕。這個ID懇求的規(guī)律分組。下面我們就來描述各種懇求和響應(yīng)消息。響應(yīng)〔Responses〕1.2.Response=>CorrelationIdResponseMessage3.4.5.CorrelationId=>int326.7.8.ResponseMessage

=> MetadataResponse |ProduceResponse

| FetchResponse

| OffsetResponse |OffsetCommitResponse|OffsetFetchResponse9.域〔FIELD〕 描述數(shù)。全部響應(yīng)都是與懇求成對匹配〔例如，我們將發(fā)送回一個元數(shù)據(jù)懇求，會得到一個元數(shù)據(jù)響應(yīng)〕。消息集〔Messagesets〕生產(chǎn)和獵取消息指令懇求共享同一個消息集構(gòu)造。在Kafka中，消息是由一個鍵值對以及少量相關(guān)的元數(shù)據(jù)組成。消息集學(xué)問一個有偏移量和大小信息的消息序列。這種格式正好即可用于在broker上的磁盤上存儲，也可用在線上數(shù)據(jù)交換。消息集也是Kafka中的壓縮單元，我們也允許消息遞歸包含壓縮消息從而允許批量壓縮。留意,在通訊協(xié)議中，消息集之前沒有類似的其他數(shù)組元素的int32。1.2.MessageSet=>[OffsetMessageSizeMessage]3.4.5.Offset=>int646.7.8.MessageSize=>int329.消息格式1.2.Message=>CrcMagicByteAttributesKeyValue3.4.5.Crc=>int326.7.8.MagicByte=>int89.10.11. Attributes=>int812.13.14. Key=>bytes8.域〔FIELD〕OffsetCrcMagicByteAttributesKeyValue

Value=>bytes描述Kafka息，實(shí)際上它并不知道偏移量的具體值，這時候它可以填寫任意值。CrcCRC32broker息的完整性。這是一個用于允許消息二進(jìn)制格式的向后兼容演化的版本id。當(dāng)0。這個字節(jié)保存有關(guān)信息的元數(shù)據(jù)屬性。最低的20。KeyKeynull.Kafkanull。壓縮〔Compression〕Kafka支持壓縮多條消息以提高效率，固然，這比壓縮一條原始消息要來得簡潔。由于單個消息可能沒有足夠的冗余信息以到達(dá)良好的壓縮比，壓縮的多條信息必需以特別方式批量發(fā)送〔固然，假設(shè)真的需要的話，你可以自己壓縮批處理的一個消息〕。要被發(fā)送的消息被包裝〔未壓縮〕在一個MessageSet構(gòu)造中，然后將其壓縮并存儲在一個單一的“消息”中，一起保存的還有相應(yīng)的壓縮編解碼集。接收系統(tǒng)通過解壓縮得到實(shí)際的消息集。外層MessageSet應(yīng)當(dāng)只包含一個壓縮的“消息”〔Kafka-1718〕?？ǚ蚩壳爸С忠幌聝煞N壓縮編解碼器編號：壓縮算法〔COMPRESSION〕編碼器編號〔CODEC〕None 0GZIP 1Snappy 2接口(TheAPIs)本節(jié)將給出每個API的用法、二進(jìn)制格式，以及它們的字段的含義的細(xì)節(jié)。元數(shù)據(jù)接口〔MetadataAPI〕這個API答復(fù)以下問題：存在哪些主題〔Topic〕？每個主題有幾個分區(qū)〔Partition〕？每個分區(qū)的Leader分別是哪個broker？這些broker的地址和端口分別是什么？這是唯一一個能發(fā)往集群中任意一個broker的懇求消息。由于可能有很多主題，客戶端可以給一個的可選主題名列表，以便只返回主題元數(shù)據(jù)的一個子集。返回的元數(shù)據(jù)是在分區(qū)級別，為了便利和以避開冗余，以主題為組集中在一起。每個分區(qū)的元數(shù)據(jù)中包含了leader以及全部副本以及正在同步的副本的信息。留意:假設(shè)broker配置中設(shè)置了”auto.create.topics.enable”,主題元數(shù)據(jù)懇求將會以默認(rèn)的復(fù)制因子和默認(rèn)的分區(qū)數(shù)為參數(shù)創(chuàng)立主題。主題元數(shù)據(jù)懇求〔TopicMetadataRequest〕1.2.TopicMetadataRequest=>[TopicName]3.4.5.TopicName=>string6.域〔FIELD〕 描述TopicName 元數(shù)據(jù)反響〔MetadataResponse〕響應(yīng)包含的每個分區(qū)的元數(shù)據(jù)，這些分區(qū)元數(shù)據(jù)以主題為組組裝brokeridbroker。每個broker一個地址和端口。1.2.MetadataResponse=>[Broker][TopicMetadata]3.4.5.Broker=>NodeIdHostPort(anynumberofbrokersmaybereturned)6.7.8.NodeId=>int329.10.11. Host=>string12.13.14. Port=>int3215.16.17. TopicMetadata => TopicErrorCode TopicName[PartitionMetadata]1.

TopicErrorCode=>int1622.23. PartitionMetadata => PartitionErrorCodePartitionIdLeaderReplicasIsr24.25.26. PartitionErrorCode=>int1627.28.29. PartitionId=>int3230.31.32. Leader=>int3233.34.35. Replicas=>[int32]9.域〔FIELD〕Leader

Isr=>[int32]描述KafkabrokeridLeaderLeaderid-1。Replicas slaveIsrBroker

〔“caughtup”，表示數(shù)據(jù)已經(jīng)完全復(fù)制到這些節(jié)點(diǎn)〕狀態(tài)的子集kafkabrokerid,主機(jī)名,端口信息可能的錯誤碼〔PossibleErrorCodes〕UnknownTopic(3)LeaderNotAvailable(5)InvalidTopic(17)TopicAuthorizationFailed(29)生產(chǎn)者接口〔ProduceAPI〕生產(chǎn)者API用于將消息集發(fā)送到效勞器。為了提高效率，它允許在單個懇求中發(fā)送多個不同主題的不同分區(qū)的消息。生產(chǎn)者API使用通用的消息集格式，但由于發(fā)送時還沒有被安排偏移量，因此可以任意填寫該值。生產(chǎn)消息懇求〔ProduceRequest〕1.2.ProduceRequest=>RequiredAcksTimeout[TopicName[PartitionMessageSetSizeMessageSet]]3.4.5.RequiredAcks=>int166.7.8.Timeout=>int329.10.11. Partition=>int3212.13.14. MessageSetSize=>int3215.域〔FIELD〕 描述這個值表示效勞端收到多少確認(rèn)后才發(fā)送反響消息給客戶RequiredAcks

0，那么效勞端將不發(fā)送反響消息〔這是唯一的效勞端不發(fā)送反響消息的狀況〕1，那么效勞器將等到數(shù)據(jù)寫入到本地日之后發(fā)送反響消息。假設(shè)這個TimeoutTopicNamePartition

值是-1，那么效勞端將堵塞，知道這個消息被全部的同步副1堵塞，直到收到這個數(shù)量的寫入反響后再反響響應(yīng)消息〔但效勞器不會等大于同步中副本的數(shù)量，即到達(dá)同步中復(fù)本個數(shù)后，會停頓等待，即使所填的值大于這個副本個數(shù)〕。這個值供給了以毫秒為單位的超時時間，效勞器可以在這個時間內(nèi)可以等待接收所需的Ack的限制，有以下緣由：〔1〕不包括網(wǎng)絡(luò)延遲，〔2〕計(jì)時器開頭在這一懇求的處理開頭，所以假設(shè)有很多懇求，由于效勞器負(fù)載而導(dǎo)致的排隊(duì)等待時間將不被包括在內(nèi)，〔3〕假設(shè)本地寫入時間超過超時，我們將不會終止本地寫操作，這樣這個超時時間就不會得到遵守。要使硬超時時間，客戶端應(yīng)當(dāng)使用套接字超時。該數(shù)據(jù)將會公布到的分區(qū)MessageSetSize后續(xù)消息集的長度，字節(jié)為單位MessageSet 上面描述的標(biāo)準(zhǔn)格式的消息集合生產(chǎn)消息響應(yīng)〔ProduceResponse〕1.2.ProduceResponse=>[TopicName[PartitionErrorCodeOffset]]3.4.5.TopicName=>string6.7.8.Partition=>int329.10.11. ErrorCode=>int1612.13.14. Offset=>int6415.域 描述Topic 假設(shè)有，此分區(qū)對應(yīng)的錯誤信息。錯誤以分區(qū)為單位供給，由于可〔Leader〕，但是其他的分區(qū)的懇求操作成功的狀況Offset 追加到該分區(qū)的消息集中的安排給第一個消息的偏移量?？赡艿腻e誤碼〔PossibleErrorCodes〕:(未完待續(xù)TODO)獵取消息接口〔FetchAPI〕獵取消息接口用于獵取一些主題分區(qū)的一個或多個的日志塊。規(guī)律上依據(jù)指定主題，分區(qū)和消息起始偏移量開頭獵取一批消息。在一般狀況下，返回消息的偏移量將大于或等于開頭偏移量。然而，假設(shè)是壓縮消息，有可能返回的消息的偏移量比起始偏移量小。這類的消息的數(shù)量通常較少，并且調(diào)用者必需負(fù)責(zé)過濾掉這些消息。獵取數(shù)據(jù)指令懇求遵循一個長輪詢模型，假設(shè)沒有足夠數(shù)量的消息可用，它們可以堵塞一段時間。作為優(yōu)化，效勞器被允許在消息集的末尾返回局部消息?？蛻魬?yīng)處理這種狀況。有一點(diǎn)要留意的是，獵取消息API需要指定消費(fèi)的分區(qū)?，F(xiàn)在的問題是如何讓消費(fèi)者知道消費(fèi)哪個分區(qū)？特別地，作為一組消費(fèi)者，如何使得每個消費(fèi)者獵取分區(qū)的一個子集，并且平衡這些分區(qū)。我們zookeeperScalaJava法的缺點(diǎn)是，它需要一個相當(dāng)胖的客戶端并且需要客戶端與zookeeperKafka〔API〕，允許該功能被移動到在效勞器端并被更便利地訪問。一個簡潔的消費(fèi)者的客戶端可以通過配置指定訪問的分區(qū)，但這樣將不能在某些消費(fèi)者失效后做到分區(qū)的動態(tài)重安排。我們期望能在下一個主要版本解決這一空白。數(shù)據(jù)獵取懇求〔FetchRequest〕1.2.FetchRequest => ReplicaId MaxWaitTime MinBytes[TopicName[PartitionFetchOffsetMaxBytes]]3.4.5.ReplicaId=>int326.7.8.MaxWaitTime=>int329.10.11. MinBytes=>int3212.13.14. TopicName=>string15.16.17. Partition=>int3218.19.20. FetchOffset=>int644.

MaxBytes=>int32域 描述IDID。一般消費(fèi)者客戶端應(yīng)ReplicaId

該始終將其指定為-1，IDbroker他們自己的節(jié)點(diǎn)ID?；谡{(diào)試目的，以非代理身份模擬副本broker-2。位。返回響應(yīng)消息的最小字節(jié)數(shù)目，必需設(shè)置。假設(shè)客戶端將此值設(shè)為0，效勞器將會馬上返回，但假設(shè)沒有的數(shù)據(jù)，效勞端會返回一個空消息集。假設(shè)它被設(shè)置為1，則效勞器將在至少一個分區(qū)收到一個字節(jié)的數(shù)據(jù)的狀況下馬上返回，或者等到超時時間達(dá)MinBytes 到。通過設(shè)置較高的值，結(jié)合超時設(shè)置，消費(fèi)者可以在犧牲一點(diǎn)實(shí)時性能的狀況下通過一次讀取較大的字節(jié)的數(shù)據(jù)塊從而提高的吞吐量〔MaxWaitTime100MinBytes64K，將允許效勞器累積數(shù)據(jù)到達(dá)64K前等待長達(dá)100ms再響應(yīng)〕。TopicName主題〔topic〕名稱idFetchOffsetMaxBytes 息的大小。獵取消息響應(yīng)〔FetchResponse〕1.2.FetchResponse => [TopicName [Partition ErrorCodeHighwaterMarkOffsetMessageSetSizeMessageSet]]3.4.5.TopicName=>string6.7.8.Partition=>int329.10.11. ErrorCode=>int1612.13.14. HighwaterMarkOffset=>int6415.16.17. MessageSetSize=>int3218.域TopicNamePartition

描述〔Topic〕名稱。id。此分區(qū)日志中最末尾的偏移量。此信息可被客戶端用來確定后面還有多少條消息。MessageSetSizeMessageSet

此分區(qū)中消息集的字節(jié)長度此分區(qū)獵取到的消息集，格式與之前描述一樣可能的錯誤碼〔PossibleErrorCodes〕OFFSET_OUT_OF_RANGE(1)UNKNOWN_TOPIC_OR_PARTITION(3)NOT_LEADER_FOR_PARTITION(6)REPLICA_NOT_AVAILABLE(9)UNKNOWN(-1)偏移量接口〔又稱ListOffset〕〔OffsetAPI〕此API描述了一個主題分區(qū)的偏移量有效范圍。生產(chǎn)者和獵取數(shù)據(jù)API的懇求必需發(fā)送到分區(qū)Leader所在的broker上，這需要通過API響應(yīng)包含分區(qū)的起始偏移量以及“日志末端偏移量”，即，將被追加到給定分區(qū)中的下一個消息的偏移量。我們也覺得這個API是略微有點(diǎn)時髦。偏移量懇求〔OffsetRequest〕1.2.OffsetRequest=>ReplicaId[TopicName[PartitionTimeMaxNumberOfOffsets]]3.4.5.ReplicaId=>int326.7.8.TopicName=>string9.10.11. Partition=>int3212.13.14. Time=>int6415.16.17. MaxNumberOfOffsets=>int3218.域 描述-1表示offset〕；-2示獵取最早的有效偏移量。留意，由于獵取到偏移值都是降序排序，因此懇求最早Offset的懇求將總是返回一個值偏移量響應(yīng)〔OffsetResponse〕1.2.OffsetResponse=>[TopicName[PartitionOffsets]]3.4.5.PartitionOffsets=>PartitionErrorCode[Offset]6.7.8.Partition=>int329.10.11. ErrorCode=>int1612.13.14. Offset=>int6415.可能的錯誤嗎〔PossibleErrorCodes〕o UNKNOWN_TOPIC_OR_PARTITION(3)NOT_LEADER_FOR_PARTITION(6)UNKNOWN(-1)偏移量提交/獵取接口〔OffsetCommit/FetchAPI〕這些API使得偏移量的能夠集中治理。了解更多偏移量治理。依據(jù)Kafka-993的評論，直到Kafka，這些API調(diào)用無法完全正0.8.2消費(fèi)者組協(xié)調(diào)員懇求〔GroupCoordinatorRequest〕消費(fèi)者組〔ConsumerGroup〕偏移量信息，由一個特定的broker維護(hù)，這個broker稱為消費(fèi)者組協(xié)調(diào)員。即消費(fèi)者需要向從這個特定的broker提交和獵取偏移量?？梢酝ㄟ^發(fā)出一組協(xié)調(diào)員覺察懇求從而獲得當(dāng)前協(xié)調(diào)員信息。1.2.GroupCoordinatorRequest=>GroupId3.4.5.GroupId=>string6.消費(fèi)者組協(xié)調(diào)員響應(yīng)〔GroupCoordinatorResponse〕1.2.GroupCoordinatorResponse=>ErrorCodeCoordinatorIdCoordinatorHostCoordinatorPort3.4.5.ErrorCode=>int166.7.8.CoordinatorId=>int329.10.11. CoordinatorHost=>string12.13.14. CoordinatorPort=>int3215.可能的錯誤碼〔PossibleErrorCodes〕GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)GROUP_AUTHORIZATION_FAILED(30)偏移量提交懇求〔OffsetCommitRequest〕1.2.v0(在.5.OffsetCommitRequest=>ConsumerGroupId[TopicName[PartitionOffsetMetadata]]6.7.8.ConsumerGroupId=>string9.10.11. TopicName=>string12.13.14. Partition=>int3215.16.17. Offset=>int6418.19.20. Metadata=>string7.28.29. v1(在1.32. OffsetCommitRequest=>ConsumerGroupIdConsumerGroupGenerationIdConsumerId[TopicName[PartitionOffsetTimeStampMetadata]]33.34.35. ConsumerGroupId=>string36.37.38. ConsumerGroupGenerationId=>int3239.40.41. ConsumerId=>string42.43.44. TopicName=>string45.46.47. Partition=>int3248.49.50. Offset=>int6451.52.53. TimeStamp=>int6454.55.56. Metadata=>string0.61.62. v2(在0.8.363.64.65. OffsetCommitRequest => ConsumerGroupConsumerGroupGenerationId

ConsumerId

RetentionTime[TopicName[PartitionOffsetMetadata]]66.67.68. ConsumerGroupId=>string69.70.71. ConsumerGroupGenerationId=>int3272.73.74. ConsumerId=>string75.76.77. RetentionTime=>int6478.79.80. TopicName=>string81.82.83. Partition=>int3284.85.86. Offset=>int6487.88.89. Metadata=>string90.在V0和v1版本中，每個分區(qū)的時間戳作為提交時間戳定義，偏移量協(xié)調(diào)員將保存消費(fèi)者所提交的偏移量，直到當(dāng)前時間超過提交時間戳+偏移量保存時間，此偏移量保存時間在broker配置中指定；假設(shè)時間錯域沒有設(shè)值，那么broker會將此值設(shè)定為接收到提交偏移量懇求的時間，用戶可以通過設(shè)置這個提交時間戳到達(dá)延長偏移量保存時間的目的。在v2版本中，我們移除了時間戳域，但是增加了一個全局保存時間域〔詳情參見KAFKA-1634〕；broker會設(shè)置提交時間戳為接收到懇求的時間，但是提交的偏移量能被保存到提交懇求中用戶指定的保存時間，假設(shè)這個保存時間沒有設(shè)值，那么broker會使用默認(rèn)的保存時間。偏移量提交響應(yīng)〔OffsetCommitResponse〕1.2.v0,v1andv2:3.4.5.OffsetCommitResponseErrorCode]]]6.7.8.TopicName=>string9.

=> [TopicName [Partition10.11. Partition=>int3212.13.14. ErrorCode=>int1615.可能的錯誤碼〔PossibleErrorCodes〕OFFSET_METADATA_TOO_LARGE(12)GROUP_LOAD_IN_PROGRESS(14)GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)ILLEGAL_GENERATION(22)UNKNOWN_MEMBER_ID(25)REBALANCE_IN_PROGRESS(27)INVALID_COMMIT_OFFSET_SIZE(28)TOPIC_AUTHORIZATION_FAILED(29)GROUP_AUTHORIZATION_FAILED(30)偏移量獵取懇求〔OffsetFetchRequest〕依據(jù)KAFKA-1841的注釋，V0和V1是在上是一樣的，但V0〔0.8.1或更高版本支持〕從zookeeper讀取的偏移量，而V1〔0.8.2或更高版本支持〕從卡夫卡讀偏移。1.2.OffsetFetchRequest => ConsumerGroup [TopicName[Partition]]3.4.5.ConsumerGroup=>string6.7.8.TopicName=>string2.

Partition=>int32偏移量獵取響應(yīng)〔OffsetFetchResponse〕1.2.OffsetFetchResponse=>[TopicName[PartitionOffsetMetadataErrorCode]]3.4.5.TopicName=>string6.7.8.Partition=>int329.10.11. Offset=>int6412.13.14. Metadata=>string8.

ErrorCode=>int16請留意，消費(fèi)者組下一個主題的分區(qū)假設(shè)沒有偏移量，broker不會設(shè)定一個錯誤碼〔由于它不是一個真正的錯誤〕，但會返回空的元數(shù)據(jù)并將偏移字段為-1?？赡艿腻e誤碼〔PossibleErrorCodes〕UNKNOWN_TOPIC_OR_PARTITION(3)<-只在求中消滅GROUP_LOAD_IN_PROGRESS(14)NOT_COORDINATOR_FOR_GROUP(16)ILLEGAL_GENERATION(22)UNKNOWN_MEMBER_ID(25)TOPIC_AUTHORIZATION_FAILED(29)GROUP_AUTHORIZATION_FAILED(30)組籍治理接口〔GroupMembershipAPI〕

v0版本的請這些懇求用于客戶端參與卡夫卡所治理的消費(fèi)者組。從更高層次上看，集群中每個消費(fèi)者組都會安排一個broker〔及消費(fèi)者組協(xié)調(diào)員〕，以簡化消費(fèi)者組治理。一旦得到了組協(xié)調(diào)員地址〔使用上面的消費(fèi)者組協(xié)調(diào)員懇求〕，組成員可以參與該組，同步狀態(tài)，然后認(rèn)真跳消息保持在組中的活潑狀態(tài)。當(dāng)客戶端關(guān)閉時，它會使用離開組請求從消費(fèi)者組中注銷。此協(xié)議的語義在Kafka客戶端安排協(xié)議中有具體描述。組建治理接口的主要使用場景是消費(fèi)者組，但這些懇求也盡量設(shè)計(jì)得一般化以便支持其他應(yīng)用場景〔KafkaConnect組〕。這種設(shè)計(jì)的帶來的代價(jià)就是是一些特定的組語義(groupsemantics)被推到了客戶端實(shí)現(xiàn)。例如，下面定義的JoinGroup和SyncGroup懇求無明確定義的字段以支持消費(fèi)者組分區(qū)安排。相反，它們在其中包含有一些通用的字節(jié)數(shù)組〔bytearrays〕，用這些字節(jié)數(shù)組就可以使得分區(qū)安排切入在消費(fèi)者客戶端實(shí)現(xiàn)。但是，雖然這種實(shí)現(xiàn)允許每個客戶端來實(shí)現(xiàn)來定義分區(qū)方案，但是Kafka工具的兼容性要求這些客戶端使用Kafka客戶端使用的標(biāo)準(zhǔn)consumer-groups.sh這個應(yīng)用程序會假定用這種格式來顯示分區(qū)安排。因此，我們建議客戶遵循一樣的模式，使這些工具對全部客戶端實(shí)現(xiàn)都可以正常工作。參與組懇求〔JoinGroupRequest〕參與組懇求用于讓客戶端成為組的成員。當(dāng)成員參與一個現(xiàn)有組，之前參與大的全部的會員必需通過發(fā)送一個參與組的要求來重入組。當(dāng)成員第一次參與該組，成員編號將是空的〔即“”〕，但重參與的成員都應(yīng)當(dāng)使用與之前生成的一樣的會員ID。1.2.JoinGroupRequest

=> GroupId SessionTimeoutMemberIdProtocolTypeGroupProtocols3.4.5.GroupId=>string6.7.8.SessionTimeout=>int329.10.11. MemberId=>string12.13.14. ProtocolType=>string15.16.17. GroupProtocols =>ProtocolMetadata]18.19.20. ProtocolName=>string21.

[ProtocolName22.23.24.

ProtocolMetadata=>bytesProtocolType字段定義了該組實(shí)現(xiàn)的嵌入?yún)f(xié)議。組協(xié)調(diào)器確保該組中的全部成員都支持一樣的協(xié)議類型。組中包含的協(xié)議〔GroupProtocols〕字段中的協(xié)議名稱和元數(shù)據(jù)的含義取決于協(xié)議類型。請留意，參與群懇求允很多協(xié)議/元數(shù)據(jù)對。這使得滾動升級時無需停機(jī)。協(xié)調(diào)器會選擇全部成員支持的一種協(xié)議，升級后的成員既包括版本和老版本的協(xié)議，一旦全部成員都升級，協(xié)調(diào)器將選擇列在數(shù)組中最前面的組協(xié)議〔GroupProtocol〕。消費(fèi)者組:下文我們定義了消費(fèi)者組使用的嵌入?yún)f(xié)議。我們建議全部消費(fèi)者客戶端實(shí)現(xiàn)遵循這個格式，以便Kafka工具能夠?qū)θ康目蛻舳苏９ぷ?.2.ProtocolType=>“consumer“.7.8.ProtocolName=>AssignmentStrategy5.16.17.UserData18.19.

AssignmentStrategy=>stringProtocolMetadata => Version Subscription20. Version=>int1621.22.23. Subscription=>[Topic]24.25.26. Topic=>string27.28.29. UserData=>bytes30.UserData域的可以用來自定義安排策略。例如，在一個粘性分區(qū)策略實(shí)現(xiàn)中，這個字段可以包含之前的安排。在基于資源的安排策略，CPUKafkaConnect“connect”的協(xié)議類型，和協(xié)議細(xì)節(jié)也是基Connect參與組響應(yīng)〔JoinGroupResponse〕接收到來自該組中的全部成員組的參與組懇求后，協(xié)調(diào)器將選擇一個成員作為Leader，并且選擇全部成員支持的協(xié)議。Leader將收到會員的完整列表與選擇的協(xié)議相關(guān)的元數(shù)據(jù)。其他追隨者成員，會收到一個空會員數(shù)組。Leader需要檢查每個成員的元數(shù)據(jù)，并且使用下SyncGroup一旦加參與組階段完成，協(xié)調(diào)器會增加該組的GenerationId，這個Id是發(fā)送給每個成員的響應(yīng)中的一個域，同時也會在心跳和偏移量提交懇求中。當(dāng)協(xié)調(diào)器重平衡〔rebalance〕了一個組，協(xié)調(diào)器將發(fā)送一個錯誤碼，表示客戶端成員需要重參與組。假設(shè)重平衡完成前成員未重入組〔rejoin〕，那么它將有一個舊generationId，在IdILLEGAL_GENERATION1.2.JoinGroupResponse

=> ErrorCode GenerationIdGroupProtocolLeaderIdMemberIdMembers3.4.5.ErrorCode=>int166.7.8.GenerationId=>int329.10.11. GroupProtocol=>string12.13.14. LeaderId=>string15.16.17. MemberId=>string18.19.20. Members=>[MemberIdMemberMetadata]21.22.23. MemberId=>string7.

MemberMetadata=>bytes消費(fèi)者組:協(xié)調(diào)器負(fù)責(zé)選擇全部成員都兼容協(xié)議〔即分區(qū)安排策略〕，Leader是實(shí)際執(zhí)行安排的成員，參與群懇求可以包含多個安排策略，從而支持現(xiàn)有版本升級或者更改不同的安排策略?？赡艿腻e誤碼〔PossibleErrorCodes〕:GROUP_LOAD_IN_PROGRESS(14)GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)INCONSISTENT_GROUP_PROTOCOL(23)UNKNOWN_MEMBER_ID(25)INVALID_SESSION_TIMEOUT(26)GROUP_AUTHORIZATION_FAILED(30)同步組懇求〔SyncGroupRequest〕組長〔groupleader〕使用同步組懇求用來向當(dāng)前組中的全部成員進(jìn)展?fàn)顟B(tài)安排〔例如分區(qū)安排〕。全部成員參與該組后，馬上發(fā)送SyncGroupLeader1.2.SyncGroupRequest=>GroupIdGenerationIdMemberIdGroupAssignment3.4.5.GroupId=>string6.7.8.GenerationId=>int329.10.11. MemberId=>string12.13.14. GroupAssignment =>MemberAssignment]15.16.17. MemberId=>string

[MemberId1.

MemberAssignment=>bytes消費(fèi)者組:消費(fèi)則組中MemberAssignment域的格式如下:1.2.MemberAssignment=>VersionPartitionAssignment3.4.5.Version=>int166.7.8.PartitionAssignment=>[Topic[Partition]]9.10.11. Topic=>string12.13.14. Partition=>int3215.16.17. UserData=>bytes18.全部了“consumer”協(xié)議類型的客戶端實(shí)現(xiàn)都需要支持這個方案同步組響應(yīng)〔SyncGroupResponse〕組中的每個成員都會接收到leader發(fā)出的syncgroup指令Eachmemberinthegroupwillreceivetheassignmentfromtheleaderinthesyncgroupresponse.1.2.SyncGroupResponse=>ErrorCodeMemberAssignment3.4.5.ErrorCode=>int166.7.8.MemberAssignment=>bytes9.可能的錯誤代碼〔PossibleErrorCodes〕:GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)ILLEGAL_GENERATION(22)UNKNOWN_MEMBER_ID(25)REBALANCE_IN_PROGRESS(27)GROUP_AUTHORIZATION_FAILED(30)心跳懇求〔HeartbeatRequest〕每當(dāng)一個成員參與并同步完成，他將開頭發(fā)送心跳懇求使自己留在組里。當(dāng)協(xié)調(diào)器在配置的會話超時時間內(nèi)沒有他的收到心跳懇求，該成員會被踢出該組。1.2.HeartbeatRequest=>GroupIdGenerationIdMemberId3.4.5.GroupId=>string6.7.8.GenerationId=>int32.

MemberId=>string心跳響應(yīng)〔HeartbeatResponse〕1.2.HeartbeatResponse=>ErrorCode3.4.5.ErrorCode=>int166.可能的錯誤碼〔PossibleErrorCodes〕:GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)ILLEGAL_GENERATION(22)UNKNOWN_MEMBER_ID(25)REBALANCE_IN_PROGRESS(27)GROUP_AUTHORIZATION_FAILED(30)退組懇求〔LeaveGroupRequest〕當(dāng)想要離開組群時，用戶可以發(fā)送一個退組懇求。這優(yōu)先于會話超時，由于它能使該組快速再平衡，這對于消費(fèi)者而言這意味著可以用更短的時間將分區(qū)安排到一個活動的成員。1.2.LeaveGroupRequest=>GroupIdMemberId3.4.5.GroupId=>string6.7.8.MemberId=>string9.10.11. LeaveGroupResponse12.13.14. LeaveGroupResponse=>ErrorCode15.16.17. ErrorCode=>int1618.可能的錯誤代碼〔PossibleErrorCodes〕:GROUP_LOAD_IN_PROGRESS(14)CONSUMER_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_CONSUMER(16)UNKNOWN_CONSUMER_ID(25)GROUP_AUTHORIZATION_FAILED(30)治理接口〔AdministrativeAPI〕組列表懇求〔ListGroupsRequest〕該API可用于找到當(dāng)前被broker治理的組群。為了得到集群內(nèi)的broker1.2.ListGroupsRequest=>3.4.5.ListGroupsResponse6.7.8.ListGroupsResponse=>ErrorCodeGroups9.10.11. ErrorCode=>int1612.13.14. Groups=>[GroupIdProtocolType]15.16.17. GroupId=>string18.19.20. ProtocolType=>string21.可能的錯誤代碼〔PossibleErrorCodes〕:GROUP_COORDINATOR_NOT_AVAILABLE(15)AUTHORIZATION_FAILED(29)組明細(xì)懇求〔DescribeGroupsRequest〕1.2.DescribeGroupsRequest=>[GroupId]3.4.5.GroupId=>string6.組明細(xì)反響〔DescribeGroupsResponse〕1.2.DescribeGroupsResponse=>[ErrorCodeGroupIdStateProtocolTypeProtocolMembers]3.4.5.ErrorCode=>int166.7.8.GroupId=>string9.10.11. State=>string12.13.14. ProtocolType=>string15.16.17. Protocol=>string18.19.20. Members => [MemberIdMemberMetadataMemberAssignment]21.

ClientId ClientHost22.23. MemberId=>string24.25.26. ClientId=>string27.28.29. ClientHost=>string30.31.32. MemberMetadata=>bytes33.34.35. MemberAssignment=>bytes36.可能的錯誤碼〔PossibleErrorCodes〕:GROUP_LOAD_IN_PROGRESS(14)GROUP_COORDINATOR_NOT_AVAILABLE(15)NOT_COORDINATOR_FOR_GROUP(16)AUTHORIZATION_FAILED(29)常量〔Constants〕接口關(guān)鍵字〔ApiKeys〕下面是懇求中ApiKey的數(shù)字值，用來表示上面所述的懇求類型。接口名稱〔APINAME〕APIKEYProduceRequest0FetchRequest1OffsetRequest2MetadataRequest3Non-userfacingcontrolAPIs4-7OffsetCommitRequest8OffsetFetchRequest9GroupCoordinatorRequest10JoinGroupRequest11HeartbeatRequest12LeaveGroupRequest13SyncGroupRequest14DescribeGroupsRequest15ListGroupsRequest16錯誤代碼〔ErrorCodes〕我們用數(shù)字代碼表示效勞器發(fā)生的問題。這些可以由客戶端轉(zhuǎn)換成客戶端中的特別(Exceptions)或者其他任何適當(dāng)?shù)腻e誤處理機(jī)制。這里是當(dāng)前正在使用的錯誤代碼表：編碼是否可重試NoError

〔CODE〔RETRIABLDESCRIPTION 描述〕 E〕0 Noerror–itUnknown -1OffsetOutOfRange 1

worked!servererror知錯誤Therequestedoffset isoutside range offsets the forthegiven移量。topic/partition.Thisindicates

InvalidMessage /

that aCorruptMessage

Yes

messagecontentsdoesnotmatchitsCRCThisrequest

它的CRC合。UnknownTopicOrPartitionInvalidMessageSize

Yes4

is for topic partition existonthis區(qū)。broker.The has a為負(fù)數(shù)。negativesizeThiserroristhrownifweare in the會 在middleofaleader選leadership electionandLeaderNotAvailable5Yesthere iscurrentlynoleaderfor有thisleader因partitionand被henceitisunavailableforwrites.Thiserroristhrowniftheclient

attempts tosendmessages

NotLeaderForPartition

Yes

區(qū)的forsomeleader。partition.Itindicates

that theclientsmetadata outofdate.Thiserroristhrowniftherequest

過期。RequestTimedOut

Yes

exceeds theuser-specifiedtimelimitintherequest.

誤BrokerNotAvailable 8ReplicaNotAvailable 9MessageSizeTooLarge 10

具用在沒場合。當(dāng)broker時拋出略〕。The serverStaleControllerEpochCode 11OffsetMetadataTooLargeCode12GroupLoadInProgressCode 14

has maximum to unbounded memory allocation. client attempt produce a誤。messagelarger thismaximum.Internal之communication.offset串時觸metadata發(fā)。會這個錯誤：當(dāng)人Yes offsets 移量時區(qū)的發(fā)生變化后15leCode

response group membership requests 員請求(such heartbeats) when metadata by the載。coordinator.The returnsthis還沒有被活是會coordinator誤。is notactive.The brokerreturnsthiserrorcodeif

調(diào)器的NotCoordinatorForGroupCode16

it 接an Yes fetch orcommitrequestforagroupthati