Kafka通訊協(xié)議指南

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

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

=> Size (RequestMessage |描述MessageSizeMessageSize域給出了 后續(xù)懇求或響應(yīng)消息的字節(jié)(bytes)長(zhǎng)度?？蛻?hù)端可以先讀取4字節(jié)的長(zhǎng)度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〔即它表示是一個(gè)元數(shù)據(jù)懇求？生產(chǎn)懇求？獵取懇求等〕.API號(hào)，該版本號(hào)允許效勞器依據(jù)版本號(hào)正確地解釋?xiě)┣髢?nèi)容。響API0。客戶(hù)端。用于匹配客戶(hù)機(jī)和效勞器之間的懇求和響應(yīng)。ClientId

場(chǎng)景。例如，你可能不僅想要監(jiān)視每秒的總體懇求，還要依據(jù)客戶(hù)端應(yīng)用程序進(jìn)展監(jiān)視，那它就可以被用上〔其中每一個(gè)都將駐留在多個(gè)效勞器上〕。這個(gè)ID懇求的規(guī)律分組。下面我們就來(lái)描述各種懇求和響應(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)都是與懇求成對(duì)匹配〔例如，我們將發(fā)送回一個(gè)元數(shù)據(jù)懇求，會(huì)得到一個(gè)元數(shù)據(jù)響應(yīng)〕。消息集〔Messagesets〕生產(chǎn)和獵取消息指令懇求共享同一個(gè)消息集構(gòu)造。在Kafka中，消息是由一個(gè)鍵值對(duì)以及少量相關(guān)的元數(shù)據(jù)組成。消息集學(xué)問(wèn)一個(gè)有偏移量和大小信息的消息序列。這種格式正好即可用于在broker上的磁盤(pán)上存儲(chǔ)，也可用在線上數(shù)據(jù)交換。消息集也是Kafka中的壓縮單元，我們也允許消息遞歸包含壓縮消息從而允許批量壓縮。留意,在通訊協(xié)議中，消息集之前沒(méi)有類(lèi)似的其他數(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í)際上它并不知道偏移量的具體值，這時(shí)候它可以填寫(xiě)任意值。CrcCRC32broker息的完整性。這是一個(gè)用于允許消息二進(jìn)制格式的向后兼容演化的版本id。當(dāng)0。這個(gè)字節(jié)保存有關(guān)信息的元數(shù)據(jù)屬性。最低的20。KeyKeynull.Kafkanull。壓縮〔Compression〕Kafka支持壓縮多條消息以提高效率，固然，這比壓縮一條原始消息要來(lái)得簡(jiǎn)潔。由于單個(gè)消息可能沒(méi)有足夠的冗余信息以到達(dá)良好的壓縮比，壓縮的多條信息必需以特別方式批量發(fā)送〔固然，假設(shè)真的需要的話，你可以自己壓縮批處理的一個(gè)消息〕。要被發(fā)送的消息被包裝〔未壓縮〕在一個(gè)MessageSet構(gòu)造中，然后將其壓縮并存儲(chǔ)在一個(gè)單一的“消息”中，一起保存的還有相應(yīng)的壓縮編解碼集。接收系統(tǒng)通過(guò)解壓縮得到實(shí)際的消息集。外層MessageSet應(yīng)當(dāng)只包含一個(gè)壓縮的“消息”〔Kafka-1718〕。卡夫卡目前支持一下兩種壓縮編解碼器編號(hào)：壓縮算法〔COMPRESSION〕編碼器編號(hào)〔CODEC〕None 0GZIP 1Snappy 2接口(TheAPIs)本節(jié)將給出每個(gè)API的用法、二進(jìn)制格式，以及它們的字段的含義的細(xì)節(jié)。元數(shù)據(jù)接口〔MetadataAPI〕這個(gè)API答復(fù)以下問(wèn)題：存在哪些主題〔Topic〕？每個(gè)主題有幾個(gè)分區(qū)〔Partition〕？每個(gè)分區(qū)的Leader分別是哪個(gè)broker？這些broker的地址和端口分別是什么？這是唯一一個(gè)能發(fā)往集群中任意一個(gè)broker的懇求消息。由于可能有很多主題，客戶(hù)端可以給一個(gè)的可選主題名列表，以便只返回主題元數(shù)據(jù)的一個(gè)子集。返回的元數(shù)據(jù)是在分區(qū)級(jí)別，為了便利和以避開(kāi)冗余，以主題為組集中在一起。每個(gè)分區(qū)的元數(shù)據(jù)中包含了leader以及全部副本以及正在同步的副本的信息。留意:假設(shè)broker配置中設(shè)置了”auto.create.topics.enable”,主題元數(shù)據(jù)懇求將會(huì)以默認(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)包含的每個(gè)分區(qū)的元數(shù)據(jù)，這些分區(qū)元數(shù)據(jù)以主題為組組裝brokeridbroker。每個(gè)broker一個(gè)地址和端口。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ī)名,端口信息可能的錯(cuò)誤碼〔PossibleErrorCodes〕UnknownTopic(3)LeaderNotAvailable(5)InvalidTopic(17)TopicAuthorizationFailed(29)生產(chǎn)者接口〔ProduceAPI〕生產(chǎn)者API用于將消息集發(fā)送到效勞器。為了提高效率，它允許在單個(gè)懇求中發(fā)送多個(gè)不同主題的不同分區(qū)的消息。生產(chǎn)者API使用通用的消息集格式，但由于發(fā)送時(shí)還沒(méi)有被安排偏移量，因此可以任意填寫(xiě)該值。生產(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〕 描述這個(gè)值表示效勞端收到多少確認(rèn)后才發(fā)送反響消息給客戶(hù)RequiredAcks

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

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

該始終將其指定為-1，IDbroker他們自己的節(jié)點(diǎn)ID?；谡{(diào)試目的，以非代理身份模擬副本broker-2。位。返回響應(yīng)消息的最小字節(jié)數(shù)目，必需設(shè)置。假設(shè)客戶(hù)端將此值設(shè)為0，效勞器將會(huì)馬上返回，但假設(shè)沒(méi)有的數(shù)據(jù)，效勞端會(huì)返回一個(gè)空消息集。假設(shè)它被設(shè)置為1，則效勞器將在至少一個(gè)分區(qū)收到一個(gè)字節(jié)的數(shù)據(jù)的狀況下馬上返回，或者等到超時(shí)時(shí)間達(dá)MinBytes 到。通過(guò)設(shè)置較高的值，結(jié)合超時(shí)設(shè)置，消費(fèi)者可以在犧牲一點(diǎn)實(shí)時(shí)性能的狀況下通過(guò)一次讀取較大的字節(jié)的數(shù)據(jù)塊從而提高的吞吐量〔MaxWaitTime100MinBytes64K，將允許效勞器累積數(shù)據(jù)到達(dá)64K前等待長(zhǎng)達(dá)100ms再響應(yīng)〕。TopicName主題〔topic〕名稱(chēng)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〕名稱(chēng)。id。此分區(qū)日志中最末尾的偏移量。此信息可被客戶(hù)端用來(lái)確定后面還有多少條消息。MessageSetSizeMessageSet

此分區(qū)中消息集的字節(jié)長(zhǎng)度此分區(qū)獵取到的消息集，格式與之前描述一樣可能的錯(cuò)誤碼〔PossibleErrorCodes〕OFFSET_OUT_OF_RANGE(1)UNKNOWN_TOPIC_OR_PARTITION(3)NOT_LEADER_FOR_PARTITION(6)REPLICA_NOT_AVAILABLE(9)UNKNOWN(-1)偏移量接口〔又稱(chēng)ListOffset〕〔OffsetAPI〕此API描述了一個(gè)主題分區(qū)的偏移量有效范圍。生產(chǎn)者和獵取數(shù)據(jù)API的懇求必需發(fā)送到分區(qū)Leader所在的broker上，這需要通過(guò)API響應(yīng)包含分區(qū)的起始偏移量以及“日志末端偏移量”，即，將被追加到給定分區(qū)中的下一個(gè)消息的偏移量。我們也覺(jué)得這個(gè)API是略微有點(diǎn)時(shí)髦。偏移量懇求〔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的懇求將總是返回一個(gè)值偏移量響應(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.可能的錯(cuò)誤嗎〔PossibleErrorCodes〕o UNKNOWN_TOPIC_OR_PARTITION(3)NOT_LEADER_FOR_PARTITION(6)UNKNOWN(-1)偏移量提交/獵取接口〔OffsetCommit/FetchAPI〕這些API使得偏移量的能夠集中治理。了解更多偏移量治理。依據(jù)Kafka-993的評(píng)論，直到Kafka，這些API調(diào)用無(wú)法完全正0.8.2消費(fèi)者組協(xié)調(diào)員懇求〔GroupCoordinatorRequest〕消費(fèi)者組〔ConsumerGroup〕偏移量信息，由一個(gè)特定的broker維護(hù)，這個(gè)broker稱(chēng)為消費(fèi)者組協(xié)調(diào)員。即消費(fèi)者需要向從這個(gè)特定的broker提交和獵取偏移量?？梢酝ㄟ^(guò)發(fā)出一組協(xié)調(diào)員覺(jué)察懇求從而獲得當(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.可能的錯(cuò)誤碼〔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版本中，每個(gè)分區(qū)的時(shí)間戳作為提交時(shí)間戳定義，偏移量協(xié)調(diào)員將保存消費(fèi)者所提交的偏移量，直到當(dāng)前時(shí)間超過(guò)提交時(shí)間戳+偏移量保存時(shí)間，此偏移量保存時(shí)間在broker配置中指定；假設(shè)時(shí)間錯(cuò)域沒(méi)有設(shè)值，那么broker會(huì)將此值設(shè)定為接收到提交偏移量懇求的時(shí)間，用戶(hù)可以通過(guò)設(shè)置這個(gè)提交時(shí)間戳到達(dá)延長(zhǎng)偏移量保存時(shí)間的目的。在v2版本中，我們移除了時(shí)間戳域，但是增加了一個(gè)全局保存時(shí)間域〔詳情參見(jiàn)KAFKA-1634〕；broker會(huì)設(shè)置提交時(shí)間戳為接收到懇求的時(shí)間，但是提交的偏移量能被保存到提交懇求中用戶(hù)指定的保存時(shí)間，假設(shè)這個(gè)保存時(shí)間沒(méi)有設(shè)值，那么broker會(huì)使用默認(rèn)的保存時(shí)間。偏移量提交響應(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.可能的錯(cuò)誤碼〔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的注釋?zhuān)琕0和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請(qǐng)留意，消費(fèi)者組下一個(gè)主題的分區(qū)假設(shè)沒(méi)有偏移量，broker不會(huì)設(shè)定一個(gè)錯(cuò)誤碼〔由于它不是一個(gè)真正的錯(cuò)誤〕，但會(huì)返回空的元數(shù)據(jù)并將偏移字段為-1。可能的錯(cuò)誤碼〔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版本的請(qǐng)這些懇求用于客戶(hù)端參與卡夫卡所治理的消費(fèi)者組。從更高層次上看，集群中每個(gè)消費(fèi)者組都會(huì)安排一個(gè)broker〔及消費(fèi)者組協(xié)調(diào)員〕，以簡(jiǎn)化消費(fèi)者組治理。一旦得到了組協(xié)調(diào)員地址〔使用上面的消費(fèi)者組協(xié)調(diào)員懇求〕，組成員可以參與該組，同步狀態(tài)，然后認(rèn)真跳消息保持在組中的活潑狀態(tài)。當(dāng)客戶(hù)端關(guān)閉時(shí)，它會(huì)使用離開(kāi)組請(qǐng)求從消費(fèi)者組中注銷(xiāo)。此協(xié)議的語(yǔ)義在Kafka客戶(hù)端安排協(xié)議中有具體描述。組建治理接口的主要使用場(chǎng)景是消費(fèi)者組，但這些懇求也盡量設(shè)計(jì)得一般化以便支持其他應(yīng)用場(chǎng)景〔KafkaConnect組〕。這種設(shè)計(jì)的帶來(lái)的代價(jià)就是是一些特定的組語(yǔ)義(groupsemantics)被推到了客戶(hù)端實(shí)現(xiàn)。例如，下面定義的JoinGroup和SyncGroup懇求無(wú)明確定義的字段以支持消費(fèi)者組分區(qū)安排。相反，它們?cè)谄渲邪幸恍┩ㄓ玫淖止?jié)數(shù)組〔bytearrays〕，用這些字節(jié)數(shù)組就可以使得分區(qū)安排切入在消費(fèi)者客戶(hù)端實(shí)現(xiàn)。但是，雖然這種實(shí)現(xiàn)允許每個(gè)客戶(hù)端來(lái)實(shí)現(xiàn)來(lái)定義分區(qū)方案，但是Kafka工具的兼容性要求這些客戶(hù)端使用Kafka客戶(hù)端使用的標(biāo)準(zhǔn)consumer-groups.sh這個(gè)應(yīng)用程序會(huì)假定用這種格式來(lái)顯示分區(qū)安排。因此，我們建議客戶(hù)遵循一樣的模式，使這些工具對(duì)全部客戶(hù)端實(shí)現(xiàn)都可以正常工作。參與組懇求〔JoinGroupRequest〕參與組懇求用于讓客戶(hù)端成為組的成員。當(dāng)成員參與一個(gè)現(xiàn)有組，之前參與大的全部的會(huì)員必需通過(guò)發(fā)送一個(gè)參與組的要求來(lái)重入組。當(dāng)成員第一次參與該組，成員編號(hào)將是空的〔即“”〕，但重參與的成員都應(yīng)當(dāng)使用與之前生成的一樣的會(huì)員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é)議類(lèi)型。組中包含的協(xié)議〔GroupProtocols〕字段中的協(xié)議名稱(chēng)和元數(shù)據(jù)的含義取決于協(xié)議類(lèi)型。請(qǐng)留意，參與群懇求允很多協(xié)議/元數(shù)據(jù)對(duì)。這使得滾動(dòng)升級(jí)時(shí)無(wú)需停機(jī)。協(xié)調(diào)器會(huì)選擇全部成員支持的一種協(xié)議，升級(jí)后的成員既包括版本和老版本的協(xié)議，一旦全部成員都升級(jí)，協(xié)調(diào)器將選擇列在數(shù)組中最前面的組協(xié)議〔GroupProtocol〕。消費(fèi)者組:下文我們定義了消費(fèi)者組使用的嵌入?yún)f(xié)議。我們建議全部消費(fèi)者客戶(hù)端實(shí)現(xiàn)遵循這個(gè)格式，以便Kafka工具能夠?qū)θ康目蛻?hù)端正常工作1.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域的可以用來(lái)自定義安排策略。例如，在一個(gè)粘性分區(qū)策略實(shí)現(xiàn)中，這個(gè)字段可以包含之前的安排。在基于資源的安排策略，CPUKafkaConnect“connect”的協(xié)議類(lèi)型，和協(xié)議細(xì)節(jié)也是基Connect參與組響應(yīng)〔JoinGroupResponse〕接收到來(lái)自該組中的全部成員組的參與組懇求后，協(xié)調(diào)器將選擇一個(gè)成員作為L(zhǎng)eader，并且選擇全部成員支持的協(xié)議。Leader將收到會(huì)員的完整列表與選擇的協(xié)議相關(guān)的元數(shù)據(jù)。其他追隨者成員，會(huì)收到一個(gè)空會(huì)員數(shù)組。Leader需要檢查每個(gè)成員的元數(shù)據(jù)，并且使用下SyncGroup一旦加參與組階段完成，協(xié)調(diào)器會(huì)增加該組的GenerationId，這個(gè)Id是發(fā)送給每個(gè)成員的響應(yīng)中的一個(gè)域，同時(shí)也會(huì)在心跳和偏移量提交懇求中。當(dāng)協(xié)調(diào)器重平衡〔rebalance〕了一個(gè)組，協(xié)調(diào)器將發(fā)送一個(gè)錯(cuò)誤碼，表示客戶(hù)端成員需要重參與組。假設(shè)重平衡完成前成員未重入組〔rejoin〕，那么它將有一個(gè)舊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í)行安排的成員，參與群懇求可以包含多個(gè)安排策略，從而支持現(xiàn)有版本升級(jí)或者更改不同的安排策略?？赡艿腻e(cuò)誤碼〔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〕組長(zhǎng)〔groupleader〕使用同步組懇求用來(lái)向當(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é)議類(lèi)型的客戶(hù)端實(shí)現(xiàn)都需要支持這個(gè)方案同步組響應(yīng)〔SyncGroupResponse〕組中的每個(gè)成員都會(huì)接收到leader發(fā)出的syncgroup指令Eachmemberinthegroupwillreceivetheassignmentfromtheleaderinthesyncgroupresponse.1.2.SyncGroupResponse=>ErrorCodeMemberAssignment3.4.5.ErrorCode=>int166.7.8.MemberAssignment=>bytes9.可能的錯(cuò)誤代碼〔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)一個(gè)成員參與并同步完成，他將開(kāi)頭發(fā)送心跳懇求使自己留在組里。當(dāng)協(xié)調(diào)器在配置的會(huì)話超時(shí)時(shí)間內(nèi)沒(méi)有他的收到心跳懇求，該成員會(huì)被踢出該組。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.可能的錯(cuò)誤碼〔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)想要離開(kāi)組群時(shí)，用戶(hù)可以發(fā)送一個(gè)退組懇求。這優(yōu)先于會(huì)話超時(shí)，由于它能使該組快速再平衡，這對(duì)于消費(fèi)者而言這意味著可以用更短的時(shí)間將分區(qū)安排到一個(gè)活動(dòng)的成員。1.2.LeaveGroupRequest=>GroupIdMemberId3.4.5.GroupId=>string6.7.8.MemberId=>string9.10.11. LeaveGroupResponse12.13.14. LeaveGroupResponse=>ErrorCode15.16.17. ErrorCode=>int1618.可能的錯(cuò)誤代碼〔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.可能的錯(cuò)誤代碼〔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.可能的錯(cuò)誤碼〔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ù)字值，用來(lái)表示上面所述的懇求類(lèi)型。接口名稱(chēng)〔APINAME〕APIKEYProduceRequest0FetchRequest1OffsetRequest2MetadataRequest3Non-userfacingcontrolAPIs4-7OffsetCommitRequest8OffsetFetchRequest9GroupCoordinatorRequest10JoinGroupRequest11HeartbeatRequest12LeaveGroupRequest13SyncGroupRequest14DescribeGroupsRequest15ListGroupsRequest16錯(cuò)誤代碼〔ErrorCodes〕我們用數(shù)字代碼表示效勞器發(fā)生的問(wèn)題。這些可以由客戶(hù)端轉(zhuǎn)換成客戶(hù)端中的特別(Exceptions)或者其他任何適當(dāng)?shù)腻e(cuò)誤處理機(jī)制。這里是當(dāng)前正在使用的錯(cuò)誤代碼表：編碼是否可重試NoError

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

worked!servererror知錯(cuò)誤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會(huì) 在middleofaleader選leadership electionandLeaderNotAvailable5Yesthere iscurrentlynoleaderfor有thisleader因partitionand被henceitisunavailableforwrites.Thiserroristhrowniftheclient

attempts tosendmessages

NotLeaderForPartition

Yes

區(qū)的forsomeleader。partition.Itindicates

that theclientsmetadata outofdate.Thiserroristhrowniftherequest

過(guò)期。RequestTimedOut

Yes

exceeds theuser-specifiedtimelimitintherequest.

誤BrokerNotAvailable 8ReplicaNotAvailable 9MessageSizeTooLarge 10

具用在沒(méi)場(chǎng)合。當(dāng)broker時(shí)拋出略〕。The serverStaleControllerEpochCode 11OffsetMetadataTooLargeCode12GroupLoadInProgressCode 14

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

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

調(diào)器的NotCoordinatorForGroupCode16

it 接an Yes fetch orcommitrequestforagroupthati