2 posts tagged with "Kafka"

幾萬個節點之上的版本控制 · 第三場　｜　長壽的 client、滾動升級的 broker，如何協商出共同版本

本篇專注在溝通當下到底選哪個版本：一條連線怎麼決定要講第幾版的 wire protocol、metadata.version（MV）在這裡扮演什麼角色、以及協商不出版本時會看到什麼錯誤。

整個主題拆成系列，相鄰兩場各自成篇；本篇只在需要時做前情提要、不重講：

• 《Kafka 叢集版本定義與 KIP-1170》：release version（上限與規則）vs metadata.version（叢集當前啟用範圍）、MV 如何決定 record schema、feature gate 與啟動檢查。
• 《運行時的版本升降》（明彥那場）：kafka-features 的 finalize 驗證、feature↔MV 依賴與降級規則、fenced broker 擋升級等「升降當下」的行為。

正文聚焦「機制」；對應的 source code 片段與 file:line 收在文末 附錄 A：原始碼對照，想深入的人再翻。commit 基準 trunk b7b1c0a8。

Part 1 — 版本為何對不齊，又是怎麼決定的​

1. 動機：為什麼版本天生對不齊，只能連線當下協商​

大多數人談「Kafka 版本」時，腦中只有一個數字（例如 3.6、4.1），並隱含一個假設：client 與 broker 同版、一起升級——「一個版本打天下」。這個假設在單機或測試環境成立，但在生產叢集會被兩個現實打破。

第一，client 很長壽。broker 由平台團隊維護、會跟著升級；但 client 是嵌在各個應用程式裡的函式庫——某個多年前的 batch job、某個沒人敢動的 legacy service，可能到今天還抱著很舊的 kafka-client 在跑。要全公司應用同一天升 client，基本不可能。這不是隨口說的：Apache Kafka 從 0.8.0（2013）到 3.x，整整九年保留了「每一個」protocol API 版本，就因為總有舊 client 還在連；直到 4.0（KIP-896）才把 baseline 提到約 2.1（2018）。

第二，broker 逐台滾動升級，過程必然新舊並存。升 broker 是一台一台來：關一台、換 binary、起來、再下一台。過程中叢集必然「新舊 broker 混在一起」，client 也同時連到新的和舊的 broker。要不停機，就不能要求全叢集同版。

因果鏈收攏成一條：任一時刻，不同節點的 binary 能力必然不同 → 要不停機就不能鎖全叢集同一版 → 版本無法事先對齊，只能在「連線當下」由雙方決定。這就是本場主題的由來。後面會介紹三種通訊角色（client↔broker、broker↔controller、broker↔broker），先說清楚：那是「版本選擇發生的地方」，不是版本對不齊的原因——對不齊的原因就是上面兩個現實。

先預告這個選擇的帳單：把版本拆到每支 API 各自一組 [min, max] 區間，協定面積、相容性測試、非 Java client 的實作負擔，全都按 API 數放大——MongoDB 與 PostgreSQL 面對同一個問題，選的是只維護一個全域協定版本號，把演進粒度讓給實作簡單。Kafka 買到的是單支 API 獨立演進、不必整包升版；這個成本後來大到需要 KIP-482 的 flexible versions 來止血——一旦某支 API 進入 flexible version，後續的 optional tagged fields 就不必再 bump 版本。評價：對一個 client 由眾多第三方各自實作、API 數十支的生態，這筆帳划算；但它只在這種生態下划算，不是通用解。

2. 術語：講「版本」時，指的是哪一層？​

進入機制前先釐清術語。日常說的「版本」其實混了三種 scope 不同、變動時機也不同的「版本」：

release version        我這台裝了哪版 binary       per-node（ops 換 binary，逐台滾動）
metadata.version       叢集 finalized 的 feature level    cluster-wide（admin 手動 finalize，刻意跟換 binary 脫鉤）
wire protocol API ver  這條連線實際講第幾版        per-connection（runtime 每條連線各自決定）

• release version：這台節點裝了哪一版 binary。per-node，由維運換 binary、逐台滾動。
• metadata.version：叢集 finalized 的 feature level。所謂 finalize，指管理員手動宣告全叢集一致採用的 feature level；怎麼宣告是第一場《版本定義》的主題，本場只需要知道它是叢集共識的一個值。cluster-wide，刻意跟換 binary 脫鉤——升了 binary 不代表 MV 就跟著升。
• wire protocol API version：這條連線實際講第幾版。per-connection，runtime 每條連線各自決定。

這三層是三個獨立的軸，不會自動一起變：可以升了 binary（release version）卻還沒 finalize metadata.version；也可以兩個 broker binary 同版，卻因連線當下選出不同 wire protocol version 而行為不同。本場的主角是第三層——wire protocol API version。

3. 架構 (a)：三種通訊角色；要協商，先送 ApiVersionsRequest​

wire 版本的資訊來源是一次查詢：連線建立後，發起端先送一支 ApiVersionsRequest，查詢對方「你每支 API 支援哪個版本區間？」，對方回覆自己每支 API 的 [min, max] 範圍。這個機制由 KIP-35 引入，是後續一切版本選擇的前提。

這個查詢有代價：每條新連線多付一次 round trip——NetworkClient 會把新連線排進 ApiVersionsRequest 流程、收到回覆後才視為 ready（NetworkClient.java:1106 / :1122）。換到的是之後每支 request 都不必再猜版本：一次性成本，長期攤提。

Kafka 叢集裡有三種通訊角色（各列的 RPC 僅代表性、非窮舉）。絕大多數路徑連線後都先做這個查詢；唯一不查詢的是 partition replication（follower→leader，元件是 replica fetcher）那條流程——原因見下一節：

client ↔ broker       讀寫資料、查 metadata        Produce、Fetch、Metadata…
broker ↔ controller   註冊、心跳、轉發 admin 請求   BrokerRegistration、BrokerHeartbeat…
broker ↔ broker       partition replication / txn markers   Fetch、ListOffsets / WriteTxnMarkers…

注意 broker↔broker 不是鐵板一塊：partition replication（follower→leader）不查詢，但同是 broker↔broker 的 transaction markers（WriteTxnMarkers）仍照 KIP-35 協商（discoverBrokerVersions=true）。所以「不查詢」修飾的是 partition replication 這條特定流程，不是整個 broker↔broker 角色。

查詢只解決「知道對方會講什麼」；拿到區間之後，最終版本怎麼定，各路徑的答案並不相同——這是下一節的主題。

4. 架構 (b)：查詢之後，最終版本誰說了算？​

先看各路徑各自的現象：

• client ↔ broker：協商。 一般 producer / consumer / admin API，取雙方區間交集的最高版本（NodeApiVersions.latestUsableVersion）。MetadataVersion 的 javadoc 也寫明：「when communicating with clients, the client decides on the API version.」
• broker ↔ controller：也是協商。 broker 對 controller 送 BrokerHeartbeat、BrokerRegistration 等 request 時，扮演的是 client 角色，同樣取交集最高版，不由 MV 決定。
• partition replication（follower→leader）：不協商。 這條由 replica fetcher 走的流程，發起端不查詢對方版本（discoverBrokerVersions=false），版本改由 finalized MV 事先決定。這條連線上依序發三支不協商 RPC——OffsetsForLeaderEpoch（對齊）、ListOffsets（定位）、Fetch（抓資料）；代表性的 Fetch 版本由 fetchRequestVersion(MV) 直接決定，leader 不支援即失敗，沒有退讓空間。（同是 broker↔broker 的 transaction markers 則仍協商。）

三種現象收斂到同一個機制：決定權在「組出這支 request 的程式碼」宣告的允許版本範圍。送出端建 request 時，AbstractRequest.Builder 帶一個 [oldestAllowedVersion, latestAllowedVersion]；底層送出前會把它跟對方 ApiVersions 廣播的範圍取交集：

client↔broker 與 broker↔controller 的 Builder 給全範圍，落在第一格；複製面的 Fetch 被 pin 成 [MV, MV]，落在第二格。複製面另外兩支 RPC 的分工更細（ListOffsets 以 MV 為上限、OffsetsForLeaderEpoch 寫死常數），屬進階細節，完整對照表收在附錄 A3；主訊息只需要一句：複製面的 Fetch 由 MV 決定，其餘都是協商。

為什麼複製面用 MV、其餘協商？​

受眾看到「複製面不協商」的第一個問題必然是：為什麼它不像其他連線一樣協商就好？直接的答案是：協商能讓 broker 讀得懂彼此送來的 request，但「讀得懂」不等於「跨版本行為正確」。 例：KIP-903 在 Fetch v15 加入 ReplicaState（帶 broker epoch）來擋住舊 epoch 的 follower 被加進 ISR——leader 就算讀得懂舊版 Fetch（協商也降得到舊版），少了那個欄位就擋不住這件事。所以複製面要的不是「兩端能通」，而是「全叢集一致地啟用同一版行為」；finalized MV 提供的正是 operator 可控、與滾 binary 脫鉤的原子切換——一次把全叢集切到指定版本，這也是升級要分兩階段的由來。其餘連線是點對點，各自挑最好的版本即可。

再往下想一層，「為什麼這樣分」有三個理由（設計推論，但每一條都對得上已驗證的機制）：

1. Bootstrap 的雞生蛋。 finalized MV 本身存在 metadata log 裡；broker 是靠「跟 controller 抓 metadata log」才知道 MV。若抓 log 的那支 Fetch 版本要由 MV 決定，就會循環——要知道 MV 得先抓 log，要抓 log 又得先知道 MV。所以啟動期的 RPC（registration、抓 metadata log 的 Fetch）不能被 MV gate，只能用自足的 ApiVersions 協商。鐵證：KRaft 抓 metadata log 的 Fetch 走全範圍協商、不看 MV（見附錄 A4）。
2. 控制面與資料面對「一致性」的需求不同。 broker↔controller 是點對點（broker 對現任 controller），每條連線各自挑最好版本即可，不需要全叢集一致。複製資料面則需要所有 follower↔leader 講同一版——MV 就是那個「集中、一次切換」的開關。
3. 不拿權威發的值去 gate 通往權威的通道。 controller 是 MV 的來源；用「它發的 MV」去決定「連到它的那條 RPC」的版本，邏輯上不成立。

另補一個常見誤解的修正：這不是「broker 加入叢集前協商、加入後改用 MV」的階段切換。BrokerHeartbeat 每隔幾秒送一次、貫穿 broker 整個生命週期，從頭到尾都是協商。真正的區分是永久按 RPC 角色分：控制面永遠協商、複製資料面永遠由 MV 決定。

MV 集中決定的帳單，與一個反事實​

這個集中決定不是免費的，帳單有三筆：升級從一步變兩步（先滾動換 binary、再手動 finalize）；多一個會被忘記的人工步驟——忘了 finalize，replication 就一直講舊版（fetchRequestVersion 的門檻表停在舊值，upgrade guide 也明確把 finalize 列為 binary 換完之後的獨立步驟）；以及沒有退讓空間——leader 不支援 MV 指定的版本就直接失敗，不像協商還能降版。要說明的是，這個「兩階段」是 finalized MV 治理的通性、不是 replication 專屬：任何由 MV gate 的能力（含 metadata log 的 record 版本）都循同一條「先滾 binary、再 finalize」的節奏。

要理解這筆帳為什麼值得付，看反事實最快：PostgreSQL 的實體複製（physical / streaming replication）傳的是 raw WAL bytes，而 WAL 格式跨大版不相容，複製能力於是硬鎖在同一個 major——所以 HA replica 不能就地跨 major 滾動升級，只能 pg_upgrade（快但要停機）。要做近零停機的大版升級，得改搭 logical replication（decode WAL 成 row changes、獨立於實體格式，可跨 major）——blue/green、另建新 major 叢集再 failover，但那是外掛（bolt-on）而非複製層內建，且有 DDL / sequence 不複製等限制。Kafka 的兩階段升級再麻煩，換到的正是 PostgreSQL 實體複製給不了的那件事：對複製層本身做線上滾動升級。我認為這是全套設計裡最站得住的取捨。（MongoDB 的 featureCompatibilityVersion 與 MV 是同一套思路；三家對照見附錄 C。）

最後一句防混淆：feature（如 group.version、share.version）不決定 RPC 版本——只有 MV 對複製面的 Fetch / ListOffsets 這麼做；feature 與 RPC 版本是兩條正交的軸（詳見附錄 A5）。

5. 以 Fetch 為例：同一顆 broker 同時講兩個版本​

把上一節的架構落到一支具體的 RPC。同一支 Fetch API 有兩種身分，分別走 client↔broker 與 broker↔broker 兩條選版路徑：

consumer fetch   Kafka 2.4 client（validVersions 0-11）  對 4.1 broker 取交集 → Fetch v11   ← 協商
replica  fetch   follower                                 版本 = fetchRequestVersion(MV) → v17  ← 由 finalized MV 決定

於是同一顆 4.1 broker，會同時對 Kafka 2.4 老 client 協商出 Fetch v11（其 validVersions 是 0-11，取交集的最高共同版本）、對 follower 用 finalized MV 決定的 v17——同一個 release，同時存在多個 wire 版本。一個版本號根本表達不了這件事。

順帶評價這個 dual-role 設計本身：consumer 與 replica 共用一支 Fetch，是「replica 也只是 log 的讀者」這個抽象的紅利——讀取路徑、fetch session 機制都只需維護一份。帳單則是 replica 專屬語意不斷滲進共用 schema：LogStartOffset 欄位註明「只在 follower 發出時使用」（FetchRequest.json:103）；KIP-903 又為了 broker epoch 驗證在 v15 加入 ReplicaState、順勢廢棄扁平的 ReplicaId。抽象紅利先收，語意租金慢慢付。

小測驗 1：replica fetch（broker↔broker）的 Fetch 版本怎麼決定？（答案見文末 附錄 B）

Part 2 — 失敗會有什麼訊息​

本場只講「通訊當下協商不出版本」的錯誤。finalize / 升降當下的錯誤——kafka-features upgrade 的 INVALID_UPDATE_VERSION、fenced broker 擋住 feature/MV 升級、feature↔MV 依賴與降級規則——屬同系列《運行時的版本升降》那場，本場不展開，只在此點一句。

6. 沒有版本交集：client 端本地中止​

client 不能直接用自己支援的最新版 API version，因為 broker 不一定支援。規則是取交集的最高版：

chosen version = max(intersection(client allowed range, broker supported range))

當交集為空，NodeApiVersions.latestUsableVersion(...) 丟出 UnsupportedVersionException；NetworkClient 接到後跳過 socket、不送出，把 request 丟進 abortedSends，最後 producer / consumer / admin 各自在收到 response.versionMismatch() 時把錯誤交回應用層。關鍵：這類 UnsupportedVersionException 很多時候不是 broker 回來的 response，而是 client 在送出前就發現沒有可用 protocol version、本地 abort。

client allows Produce 0-13 , broker supports 0-10  -> chosen 10
client allows Produce 11-13, broker supports 0-10  -> UnsupportedVersionException（送出前中止）

7. 繞過協商、直接送出不支援的版本：broker 關閉連線​

若繞過協商、直接送出一個 broker 不支援的 API version，失敗路徑是固定的一條：broker 端 RequestContext 解析 request 失敗 → 丟出 UnsupportedVersionException → SocketServer 直接關閉連線（RequestContext.java:112、SocketServer.scala:781）。

為什麼關閉連線、而不是回一個錯誤 response？broker 其實讀得到 header（知道 apiKey 與版本），但一般 response 的序列化必須依 client 指定的 request version 進行——既然那個版本本身不受支援，broker 無法保證組出來的回應能被對方正確解讀。關閉連線是唯一可靠的動作。

唯一的例外是 ApiVersions 本身：它是 bootstrap 的逃生口——即使 client 送的 ApiVersions 版本超出 broker 支援範圍，broker 也不會關線，而是回一個 v0 的 response 帶 UNSUPPORTED_VERSION 錯誤碼與自己支援的版本範圍，讓 client 得以 recover、重新協商。

小測驗 2：client 繞過協商、直接送出 broker 不支援的版本會怎樣？（答案見文末 附錄 B）

8. 版本截斷：為什麼「升一點點」不夠​

第 6 節「交集為空」最容易被忽略的根因是——舊的 wire protocol API version 會被整個移除。每個 API 都有自己的 validVersions 範圍，而這個範圍不保證永遠從 0 開始。

Kafka 4.0 就移除了一批舊 wire API 版本，例如 FetchRequest.json 的 validVersions 已是 "4-18"（Fetch v0–v3 移除，min 升到 4）。因此若 client 太舊、只會講已被截斷的版本，就會落到交集外——協商結果直接是 no usable version。這時「再升一點點」沒用，得跨過 upgrade guide 的版本下限（升任一端到 4.0 前，另一端要 ≥ 2.1，雙向要求）。

值得停一秒看這個決策的形狀。從 0.8.0（2013）到 3.x，Kafka 保留了每一個 protocol API 版本整整九年——「相容性至上」推到極端的取捨：好處是任何老 client 永遠連得上；代價是每個舊版本都是活的程式碼路徑與測試矩陣，broker 永遠不能假設 client 具備任何新能力。4.0 用 KIP-896 把 baseline 收到 2.1（2018），本質是一次帳務結算：用「斷掉 2018 年以前的 client」換「刪碼、縮測試面、讓協定假設前進七年」。為什麼等這麼久？因為斷 client 是不可逆的破壞性變更，只有 major 版本邊界的社會契約付得起。我的評價：收得對，甚至偏晚——九年的窗說明這個專案在相容性上保守到近乎自虐，而這份保守正是理解 Kafka 一切版本設計的鑰匙。

Recap​

本場的因果鏈只有一條：client 長壽、broker 滾動升級 → 版本天生對不齊 → 只能在連線當下決定每條連線講第幾版——client↔broker 與 broker↔controller 靠 ApiVersionsRequest 查詢後取交集協商，partition replication 的 Fetch 由 finalized MV 集中決定。「一個版本打天下」不成立的最好證據，就是那顆 4.1 broker：同一支 Fetch，同一時刻，對 Kafka 2.4 老 client 講 v11、對 follower 講 v17。而當版本選不出來：交集為空時 client 在送出前本地中止；繞過協商直接送出不支援的版本，broker 解析失敗、關閉連線。至於 finalize / 升降當下會出什麼錯，交給同系列《運行時的版本升降》。

最後一句立場：per-API 細粒度協商＋replication 集中治理，這套組合是為「client 生態極度分散、API 數十支」的系統量身打造的取捨——對 Kafka 划算，但不是通用解。MongoDB 選了較粗的全域粒度、PostgreSQL 乾脆放棄複製層的版本治理，各自都對得上自家的生態；讀懂一個系統的版本設計，就是讀懂它對自己使用者的假設。

附錄 A：原始碼對照​

正文把機制講完，這裡放對應的 source 片段與 file:line。commit 基準 trunk b7b1c0a8。

A1 — Builder 版本範圍與協商​

• AbstractRequest.Builder 帶 [oldestAllowedVersion, latestAllowedVersion]，NetworkClient 送出前取交集。
• 四條路徑的發起端共用同一顆 org.apache.kafka.clients.NetworkClient、各帶一份 ApiVersions cache——是否先送 ApiVersionsRequest 由建構參數 discoverBrokerVersions 決定：client、controller、KRaft 路徑為 true，replica fetcher 為 false（實作細節，正文不展開）。
• client↔broker 取交集最高版：NodeApiVersions.latestUsableVersion(...)（clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java:149）。
• Doc：server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:31「when communicating with clients, the client decides on the API version.」

A2 — broker ↔ controller（broker 當 client）​

• BrokerLifecycleManager.java:580：channelManager.sendRequest(new BrokerHeartbeatRequest.Builder(data), handler)。
• BrokerHeartbeatRequest.Builder 只有 super(ApiKeys.BROKER_HEARTBEAT)、未 pin 版本 → 全範圍協商。
• 底層：NodeToControllerChannelManagerImpl.java:67（private final ApiVersions apiVersions）、:115 / :129（NetworkClient 吃 apiVersions）。
• 走這條管線的 request：BrokerHeartbeat、BrokerRegistration、ControllerRegistration、AssignReplicasToDirs。

A3 — broker ↔ broker（複製面）三層分工​

正文只講「複製面的 Fetch 由 MV 決定，其餘都是協商」；完整版是三支 RPC 三種作法（core/src/main/scala/kafka/server/RemoteLeaderEndPoint.scala）。先補一個名詞：ListOffsets = 把時間戳／哨兵（earliest / latest / by-timestamp）換算成一個 offset（consumer 的 seekToBeginning、offsetsForTimes 靠它，follower 則用來找截斷點）。

• 底層 BrokerBlockingSender.scala:82 仍是 NetworkClient，但 :95 的 discoverBrokerVersions=false——不送 ApiVersionsRequest，直接照 MV 決定的版本送。
• MetadataVersion 裡跟 RPC 版本有關的方法只有兩個：fetchRequestVersion()（:273）與 listOffsetRequestVersion()（:289），只作用在複製面的 Fetch / ListOffsets。
• 一句解讀：同一條複製路徑上三支 RPC 三種選版法，並非設計失誤。OffsetsForLeaderEpoch 的 v4 固定值是移除舊 MetadataVersion 之後留下的簡化（KAFKA-18465 清理的結果），程式碼並明示：未來若要加新版本，須改用 metadata.version gate（OffsetsForLeaderEpochRequest.java:64 的註解）。

MetadataVersion.fetchRequestVersion()（server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:273）：

public short fetchRequestVersion() {
    if (isAtLeast(IBP_4_1_IV1)) {
        return 18;
    } else if (isAtLeast(IBP_3_9_IV0)) {
        return 17;
    } else if (isAtLeast(IBP_3_7_IV4)) {
        return 16;
    } else if (isAtLeast(IBP_3_5_IV1)) {
        return 15;
    } else if (isAtLeast(IBP_3_5_IV0)) {
        return 14;
    } else {
        return 13;
    }
}

A4 — KRaft metadata-log Fetch：協商 + 獨立 feature（進階）​

正文第 4 節「雞生蛋」的鐵證在此。broker 以 observer、controller quorum 彼此之間抓 metadata log 的 Fetch，走協商、不看 MV。

• KafkaRaftClient.buildFetchRequest()（:2985）→ RaftUtil.singletonFetchRequest(...) → KafkaNetworkChannel.buildRequest（:192）包成 FetchRequest.SimpleBuilder。
• SimpleBuilder = super(ApiKeys.FETCH) 全範圍（FetchRequest.java:133）→ ApiVersions 協商，不是 MV。
• 這條路徑的能力由獨立於 MV 的 kraft.version feature 治理（KRaftVersion.java）；KafkaRaftClient.java:185 的 localSupportedKRaftVersion: SupportedVersionRange 是各節點自報的支援範圍。

A5 — feature 與 RPC 正交、honor（進階）​

正文只留了一句「feature 不決定 RPC 版本」；完整版如下。這段與《版本定義》場的 feature gate 相鄰，放附錄避免搶正文主線。

RPC 版本與 feature 是兩條正交的軸，要用一個功能得兩個都滿足：

• RPC 支援（wire 能力）：兩端「能不能講」這支 RPC / 這個版本——由 binary 能力 + ApiVersions 協商決定（複製面由 MV）。
• feature（叢集政策）：叢集「准不准用」這個功能——由 metadata log 裡 finalized 的 feature level 決定。

要點：

• feature 不決定 RPC 版本。只有 MV 對複製面的 Fetch / ListOffsets 這麼做；feature 產出的是布林能力或 record 格式版本。
• feature 真正 gate 的是 broker 要不要 honor 一支 RPC。 honor 指 broker 收到請求後「認這筆請求、照該功能的語意去處理」，而不是拒絕、回錯或忽略。RPC 在 binary 裡一直存在、也能協商成功送達，但 broker 會查 finalized feature 決定是否 honor：group.version（KIP-848）未開時，handleConsumerGroupHeartbeat 直接 fail（KafkaApis.scala:2642-2650）；share.version 未開時不受理 ShareFetch，toggle off 時清掉 share session（SharePartitionManager.java:633、KafkaApis.scala:4290）。
• 對外 feature 需要 RPC 承載：新功能通常帶新 RPC（如 ConsumerGroupHeartbeat、ShareFetch / ShareAcknowledge）或現有 RPC 的更高版本。純內部 feature 不需要對外 RPC（只改 record 格式或內部行為）。
• 帶版本的 feature 反向用 RPC 版本推能力：transactionVersionForAddPartitionsToTxn(request) 看 request 版本 > 3 → client 支援 TV2（TransactionVersion.java:67）。方向與「feature → RPC 版本」相反。

A6 — 失敗路徑​

client 端取交集（clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java:149）：

Optional<ApiVersion> intersectVersion = ApiVersionsResponse.intersect(supportedVersion,
    new ApiVersion()
        .setApiKey(apiKey.id)
        .setMinVersion(oldestAllowedVersion)
        .setMaxVersion(latestAllowedVersion));

if (intersectVersion.isPresent())
    return intersectVersion.get().maxVersion();
else
    throw new UnsupportedVersionException(...);

NetworkClient.doSend(...) 接到 UnsupportedVersionException 後跳過 socket、丟進 abortedSends（NetworkClient.java:591 呼叫、:597 catch）；NetworkClient.poll(...) drains aborted sends（:651 / :940）。Producer / Consumer 收到 response.versionMismatch()（Sender.java:595、ConsumerNetworkClient.java:614）。

broker 端收到不支援版本的路徑：RequestContext 解析 request 失敗、丟 UnsupportedVersionException（clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:112）→ SocketServer 關閉連線（core/src/main/scala/kafka/network/SocketServer.scala:781）。

broker 組 ApiVersionsResponse 時放進每個 API 的 min/max（clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java:287）；ApiVersions 對外仍宣告 v0 的特例讓它成為 bootstrap 逃生口。版本截斷的 schema 證據：FetchRequest.json:61 = "4-18"、ListOffsetsRequest.json:45 = "1-11"、ProduceRequest.json = "3-13"。upgrade guide 對 4.0 截斷的說明：docs/getting-started/upgrade.md:229。

附錄 B：常見誤解 / 隨堂考​

兩題對應兩段主線，每題附「直覺答案（多半錯）」與正解，可當現場有獎徵答。

Q1：replica fetch（broker↔broker）的 Fetch 版本怎麼決定？​

• 直覺：兩個 broker 用 ApiVersions 協商取交集。
• 正解：不協商。由 finalized metadata.version 決定（fetchRequestVersion(MV)，建成 [v, v] exact pin）；所有 broker 從同一個 MV 推出同一版。
• 出處：RemoteLeaderEndPoint.scala:215、MetadataVersion.java:273

Q2：client 繞過協商、直接送出 broker 不支援的 API version，會怎樣？​

• 直覺：broker 一律回一個 UNSUPPORTED_VERSION 錯誤碼。
• 正解：一般 API → RequestContext 解析失敗、broker 丟 UnsupportedVersionException，SocketServer 關閉連線。唯一例外是 ApiVersions：它是 bootstrap 逃生口，會回 v0 response 帶 UNSUPPORTED_VERSION 錯誤碼 + 支援範圍讓 client recover。
• 出處：RequestContext.java:112、SocketServer.scala:781、ApiKeys.java（ApiVersions 特例）

附錄 C：別家怎麼答同一題​

完整對照（含來源 URL）的來源 URL 見各家官方文件；這裡只留骨架。

兩個讀法：MongoDB 的 FCV 證明「叢集能力世代跟 binary 脫鉤、手動 finalize」不是 Kafka 的怪癖，而是分散式系統滾動升級的共同答案；PostgreSQL 則是反事實——複製層本身不做版本治理，實體複製鎖 major，跨大版的近零停機升級只能靠外掛的 logical replication，而非複製層內建。

本篇搭配的範例程式已一併放進這個 repo：examples/kafka/kip-1034-dlq-blog-post/。

before/ 對應 Kafka 4.2.0 以前常見的手動 DLQ 作法，after/ 對應 Kafka 4.2.0 / KIP-1034 的內建 DLQ 作法。

Kafka Streams 應用在處理資料流時，經常需要面對不合法或無法反序列化的 record。遇到這類資料時，系統可以選擇 fail，讓應用停止；也可以選擇 continue，略過該筆資料。實務上通常還有第三種需求：保留錯誤資料，寫入另一個 topic，供後續修補、重送、追查或告警使用。這正是 DLQ（Dead Letter Queue）的用途。

問題在於，Kafka Streams 4.2.0 以前並未提供完整的內建 DLQ 寫入路徑。應用程式可以自行補上這項能力，但必須同時處理 producer lifecycle、error metadata、錯誤發生位置，以及 transaction 邊界；若啟用 exactly_once_v2，這些限制會更加明顯。

KIP-1034 補上的正是這段缺口。Kafka Streams 的內建 exception handlers 現在可以把 DLQ record 回交給框架，由框架透過既有寫入流程送出，使 DLQ 寫入得以回到 Kafka Streams 的 transactional write flow。以下以 repo 內的範例對照舊作法與新作法。

先界定問題​

範例使用一個 click-events topic，內容為 JSON 字串：

{"ad_id":"banner-A","count":3}

Streams topology 會把它讀進來、deserialize 成 ClickEvent，接著做一點簡單轉換，最後寫到 click-events-output。

整體流程可以想成：

click-events -> deserialize -> process -> click-events-output

若資料格式正確，處理流程相當單純。但只要出現 NOT_VALID_JSON 這類資料，Kafka Streams 在進入 processor 前就可能失敗，因為 record 必須先由 bytes 轉為應用程式期待的物件；一旦反序列化失敗，後續 processor 尚未開始執行。

因此，在 Kafka Streams 中討論 DLQ，不能只理解為「catch exception 後寫入另一個 topic」。錯誤可能發生在 topology 之內，也可能發生在 topology 之前。

case A: processing error

click-events -> deserialize -> process X -> output
                                |
                                +-> still inside topology


case B: deserialization error

click-events -> deserialize X -> process -> output
                  |
                  +-> processor 還沒開始

Kafka 4.2.0 以前：兩種 exception，兩套處理方式​

先看 Kafka 4.2.0 以前常見的手動 DLQ 作法。主要負擔在於，應用程式不是只需處理單一錯誤型態，而是必須分別處理兩種不同型態的 exception。repo 裡的 before/ 將這兩種情況整理成可重現的範例，方便對照各自限制。

Processing error：發生在 topology 裡​

若錯誤出現在 topology 內部，應用程式仍有機會在 DSL 轉換或 processor 中處理。這裡示範的做法，是先讓資料正常完成 deserialization，接著在 flatMap 這類 DSL 轉換中執行 business rule，並於必要時自行 try/catch。

這段是 ClickEventManualDlqTopology.java 的核心：

stream
    .flatMap((key, event) -> {
        try {
            if (event.count < 0) {
                throw new IllegalArgumentException("count must be non-negative");
            }
            String processed = "user=" + key + " clicked ad=" + event.adId + " count=" + event.count;
            return Collections.singletonList(KeyValue.pair(key, processed));
        } catch (Exception e) {
            sendToDlq(key, event, e);
            return Collections.emptyList();
        }
    })
    .to(outputTopic);

這段 code 的重點不在 flatMap，而在 catch 裡的 sendToDlq()。這裡的錯誤已經不是 JSON parse 失敗，而是 record 成功進入 topology 後，因 business rule 不合法而被手動導向 DLQ。舊版最常見的做法，是準備一個獨立的 KafkaProducer，遇到這類錯誤就直接寫入 DLQ。

這個方式能夠運作，也可以一併補上 error metadata：

ProducerRecord<String, String> dlqRecord = new ProducerRecord<>(dlqTopic, key, value);
dlqRecord.headers().add("error.message", cause.getMessage() != null
        ? cause.getMessage().getBytes() : "null".getBytes());
dlqRecord.headers().add("error.class", cause.getClass().getName().getBytes());
dlqProducer.send(dlqRecord).get();

限制在於，這個 dlqProducer 與 Kafka Streams 內部 producer 並非同一個 instance。應用程式不只要另行維護 producer，也無法把這條 DLQ 寫入納入 Kafka Streams 的同一個 transaction，因此 DLQ 寫入與主流程無法共同達成 EOS。

Deserialization error：發生在 topology 之前​

deserialization error 的限制更明確。

這類錯誤不是發生在 map、flatMap、transform 等 topology 步驟內，而是在 record 被 consumer 取回後、真正進入 topology 前就已發生。換言之，processor 尚未接手該筆資料，deserialization 已經失敗。

在這種情況下，topology 內的分流手段都無法介入。flatMap、split()、branch() 均無法觸及該筆資料；可用的處理入口只剩 DeserializationExceptionHandler。

ManualDlqHandler.java 示範的就是這條路：

@Override
public DeserializationHandlerResponse handle(
        ErrorHandlerContext context,
        ConsumerRecord<byte[], byte[]> record,
        Exception exception) {

    sendToDlq(record, exception);
    return DeserializationHandlerResponse.CONTINUE;
}

此處的限制相當直接：deserialization error 發生時，record 尚未進入 topology，因此不能用 topology 內的 routing 手段處理。context.forward() 無法使用，branch()、split()、flatMap 等 DSL 轉換也無法觸及該筆資料。若要寫入 DLQ，舊版常見作法仍是自行建立獨立的 KafkaProducer。

除了自行送出之外，應用程式也必須自行複製原始 headers，並補上 topic / partition / offset / exception 等 metadata：

record.headers().forEach(h -> dlqRecord.headers().add(h));
dlqRecord.headers().add("__manual.error.topic", record.topic().getBytes());
dlqRecord.headers().add("__manual.error.partition",
        String.valueOf(record.partition()).getBytes());
dlqRecord.headers().add("__manual.error.offset",
        String.valueOf(record.offset()).getBytes());

這正是舊版作法的主要負擔：不同錯誤類型必須掛在不同處理位置，處理方式也不一致。

綜合 before/ 的兩條路徑，application 層必須自行承擔下列責任：

• 判斷錯誤應於哪一層攔截；processing error 與 deserialization error 並非同一套寫法。
• 另行維護 KafkaProducer，包含 lifecycle、配置與送出失敗時的處理策略。
• 自行定義 headers 命名與需要攜帶的 metadata。
• 在 EOS 開啟時承擔 DLQ 可能於 retry 過程中重複寫入的風險。
• 測試往往必須圍繞 workaround 撰寫，而不是直接驗證框架行為。

舊版的問題不只是程式碼較多，而是錯誤處理、資料一致性與 observability 的責任都被推回 application 層。

真正的痛點在 tx-safe​

即使接受「只能自行寫入 DLQ」這個前提，transaction 邊界仍然沒有解決。

Kafka Streams 在 EOS 模式下會自行管理 transaction。簡化後的處理流程如下：

BEGIN TX
  -> consume record
  -> deserialize
  -> process
  -> send output via RecordCollector
  -> sendOffsetsToTransaction
COMMIT TX

若在此流程中另以獨立 KafkaProducer 寫入 DLQ，問題相當直接：dlqProducer 與 Kafka Streams 內部用於送出 output 的 producer 不是同一個 producer instance。既然不是同一個 producer，就無法共享同一個 Kafka transaction。

用圖看會更清楚：

Kafka Streams internal producer
  -> output records
  -> transaction A

manual dlqProducer
  -> DLQ records
  -> not part of transaction A

也就是說，手動送出的 DLQ record 不會落在 Kafka Streams 那條 transactional write path 裡。

可能發生的情境如下：

1. 手動 DLQ producer 已經將錯誤資料送出。
2. Kafka Streams 內部 transaction 隨後因 rebalance、crash 或其他原因 abort。
3. Kafka Streams retry 後重新處理同一筆資料。
4. DLQ record 再次被寫入。

Kafka 4.2.0 / KIP-1034：框架終於把這條路補起來​

到了 Kafka 4.2.0，KIP-1034 將這件事正式納入 Kafka Streams 的 error handling flow。

核心方向是：exception handler 可以把要寫入 DLQ 的 records 回交給框架，由 Kafka Streams 透過內部 producer 送出。這項能力加在 Kafka Streams 內建的 deserialization / processing / production exception handling 流程上；本文的 after/ 範例同時示範 deserialization error 與 processing error。

這個改變的關鍵在於，只要 DLQ record 由框架送出，就能沿用 Kafka Streams 既有的 producer 與 transaction，而不必在 application 層另行建立獨立 producer。

KIP-1034 之後，topology 本身可以維持單純：

builder
    .stream(inputTopic, Consumed.with(Serdes.String(), new ClickEventSerde()))
    .mapValues(event -> {
        if (event.count < 0) {
            throw new IllegalArgumentException("count must be non-negative");
        }
        return "user clicked ad=" + event.adId + " count=" + event.count;
    })
    .to(outputTopic);

這段 ClickEventTopology.java 仍不包含任何 DLQ 相關 code；沒有手動 try/catch、沒有另行建立 producer，也沒有自行補 headers。count < 0 只是 business validation；該 exception 如何寫入 DLQ，仍由 Kafka Streams 的 processing exception handler 負責。

真正啟用 DLQ 的設定位於 App.java。DLQ topic 由下列 config 指定：

props.put(StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG, DLQ_TOPIC);

再搭配內建 deserialization / processing handlers：

props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
        LogAndContinueExceptionHandler.class);
props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
        LogAndContinueProcessingExceptionHandler.class);

這裡也是 KIP-1034 最核心的 API 變化。舊版 exception handler 的回傳值本質上只是在回答「繼續」或「失敗」；4.2.0 之後，handler 的新 Response 可以額外攜帶「需要由框架送出的 DLQ records」。也因為 handler 現在可以把 DLQ records 回交給 Kafka Streams，框架才得以透過內部 StreamsProducer / RecordCollector 送出，而不是把 DLQ 寫入責任留在 application 層。

本文範例中的 malformed JSON 會觸發 deserialization error，因此 LogAndContinueExceptionHandler 這條路徑的效果如下：

1. ClickEventSerde 反序列化失敗時，會拋 exception。
2. LogAndContinueExceptionHandler 會接手。
3. 4.2.0 的 handler 可以建立 DLQ record，並交還給 Kafka Streams。
4. Kafka Streams 透過 RecordCollectorImpl 用同一個 StreamsProducer 把 record 送出去。
5. 因為走的是同一個 producer，DLQ 寫入也落在同一個 transaction 邊界內。

本文範例中的 count < 0 則會觸發 topology 內部的 processing error，由 LogAndContinueProcessingExceptionHandler 接手。Kafka Streams 4.2.0 的 processing.exception.handler 預設值是 LogAndFailProcessingExceptionHandler；即使設定了 errors.dead.letter.queue.topic.name，預設 handler 仍會回傳 fail。因此，若 processing error 也要「寫入 DLQ 後繼續」，必須明確設定 LogAndContinueProcessingExceptionHandler。

補充一點：ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG 之所以有效，是因為內建 exception handlers 會讀取該 config，並透過 Kafka Streams 內部工具建立 DLQ record；它不是框架對所有 handler 強制套用的行為。多數情境下，內建 handler 已足以涵蓋需求；若需要自訂 DLQ record 內容，仍可改用 custom handler。

但 custom handler 不必回到舊版的手動 producer 寫法。KIP-1034 之後，exception handler 介面本身已改變：舊版 handle() 只能回傳 CONTINUE 或 FAIL；新版 handleError() 回傳 Response，其中可以攜帶 ProducerRecord 列表，由 Kafka Streams 透過同一個內部 producer 送出。

// Kafka 4.2.0 以前：若要寫入 DLQ，通常只能自行建立 producer
@Override
public DeserializationHandlerResponse handle(
        ErrorHandlerContext context,
        ConsumerRecord<byte[], byte[]> record,
        Exception exception) {
    ProducerRecord<byte[], byte[]> dlqRecord =
            new ProducerRecord<>("app-dlq", record.key(), record.value());
    dlqProducer.send(dlqRecord).get();  // 獨立 producer，不在 Streams tx 裡
    return DeserializationHandlerResponse.CONTINUE;
}

// Kafka 4.2.0：把 DLQ record 回交給框架，走同一條 transaction 路徑
@Override
public DeserializationExceptionHandler.Response handleError(
        ErrorHandlerContext context,
        ConsumerRecord<byte[], byte[]> record,
        Exception exception) {
    ProducerRecord<byte[], byte[]> dlqRecord =
            new ProducerRecord<>("app-dlq", record.key(), record.value());
    return DeserializationExceptionHandler.Response.resume(List.of(dlqRecord));
}

介面差異就在於：handleError() 允許 handler 把 DLQ records 回交給框架送出，不需要在 application 層另行建立 producer。

這是 KIP-1034 最重要的差別。它不只是省去自行建立 producer 的負擔，也讓 DLQ 寫入重新納入 Kafka Streams 的一致性模型。

換言之，本文所說的「透過 config 啟用 DLQ」成立的前提，是使用內建 handler；若改用 custom deserialization / processing / production handler，errors.dead.letter.queue.topic.name 不會自動替該 handler 建立 DLQ record，handler 必須自行決定是否建立 records。不過，custom handler 仍可透過 Response.resume(...) 把 records 回交給 Kafka Streams，因此依然可以走內建寫入路徑，而不需要自行建立 producer。

若拆開 after 帶來的差異，可以整理為下列幾點：

• 不必另行建立 DLQ producer；Kafka Streams 內部會負責送出。
• 不必把 DLQ 邏輯放入 topology；主流程可以維持單純。
• 不必自行補齊常見 error headers；exception / topic / partition / offset 等 metadata 由框架建立。
• deserialization error 不再需要 application 層 workaround；框架已提供正式處理路徑。
• processing error 也可以透過內建 handler 回交 DLQ records，不必在 topology 中手動建立 producer。
• KIP-1034 的能力也延伸到 production exception handler；ProductionExceptionHandler 處理的是 Kafka Streams 送出到下游時的寫入錯誤。4.2.0 以前，handle() 只能回傳 CONTINUE 或 FAIL，沒有 DLQ records；4.2.0 之後，handleError() 回傳 Response，可以攜帶 DLQ records，也具備 RETRY 選項。
• DLQ 與 exactly_once_v2 可以放在同一個 transaction 模型中理解。

整體而言，設定更集中，topology 更單純，error metadata 由框架補齊，測試也能直接驗證框架行為。更重要的是，DLQ 寫入回到 transaction 邊界內，能與 EOS 模型一致。

after 的資料流則比較像這樣：

                +-------------------+
                |   click-events    |
                +-------------------+
                          |
                          v
                +-------------------+
                |   deserialize     |
                +-------------------+
                    |           |
             success|           |error
                    v           v
             +-------------+   +------------------------------+
             |   process   |   | LogAndContinueExceptionHandler|
             +-------------+   +------------------------------+
                    |                         |
                    v                         v
          +-------------------+     +-------------------+
          | click-events-out  |     | click-events-dlq  |
          +-------------------+     +-------------------+

          both writes go through Kafka Streams internals

如果只看 transaction 邊界，after 的差異會更明顯：

+---------------- Kafka Streams transaction ----------------+
| consume -> deserialize -> process                        |
|                    |                                     |
|                    +-> DLQ record via Kafka Streams     |
|                    +-> output record via Kafka Streams  |
|                                                           |
| both use the same StreamsProducer / RecordCollector       |
+--------------------- commit / abort ---------------------+

不只 tx-safe，連 error headers 也一起內建​

另一項實用的改變，是 error metadata 不必再由 application 層自行補齊。

在舊版手動作法中，header 名稱、內容格式，以及是否攜帶原始 topic / partition / offset，都必須由應用程式自行決定。不同團隊可能各自定義一套格式，時間一久也容易分歧。

KIP-1034 之後，框架會自動附上這些 __streams.errors.* headers：

• __streams.errors.exception
• __streams.errors.message
• __streams.errors.stacktrace
• __streams.errors.topic
• __streams.errors.partition
• __streams.errors.offset

這個細節之所以重要，是因為 DLQ 並不只是承接錯誤資料。實務上，後續往往還牽涉到原因追查、告警、資料回補與 replay。若這些 metadata 能由框架穩定補齊，後續處理會更一致，也更容易被測試覆蓋。

總結​

Kafka 4.2.0 以前的 DLQ 較像 application 層自行補出的機制；KIP-1034 之後，DLQ 才正式進入 Kafka Streams 框架，並能與 transaction 模型一併運作。

用測試看行為差異，比講概念更準​

許多細節放在測試中觀察會更直接，因為 input、output、DLQ record 與 headers 都能在同一處驗證。若要對照本文提到的 routing 與 header 行為，可直接閱讀 before/ 與 after/ 的測試。

若要自行執行範例，專案目錄如下：

cd examples/kafka/kip-1034-dlq-blog-post
./gradlew test

何時應評估升級到 4.2.0？​

若系統具備下列需求，KIP-1034 的價值會相當明確：

• deserialization error 需要穩定寫入 DLQ。
• 不希望在 topology 外另行維護 producer。
• 已啟用 exactly_once_v2，且不希望承擔 transaction 外寫入導致的重複寫入風險。
• 需要一致的 error headers，以支援後續告警、查錯與回補。

換言之，4.2.0 之後，DLQ 不再只是 application 層自行補上的機制，而是 Kafka Streams 框架正式承接的責任。

結語​

KIP-1034 補的是 Kafka Streams 長期存在的 DLQ 缺口。

Kafka 4.2.0 以前，真正制約 DLQ 設計的是 transaction 邊界：只要使用獨立 producer 寫入 DLQ，該筆寫入就脫離 Kafka Streams 的 transaction 範圍；在 EOS 開啟時，retry 可能造成重複寫入。除此之外，錯誤攔截層級、送出方式與 headers 命名也都必須由 application 層自行維護。

4.2.0 之後，設定 StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG 並搭配內建 exception handler，錯誤資料即可交由 Kafka Streams 內部寫入 DLQ，不必另行建立 producer。對於仍在維護手動 DLQ 的 Kafka Streams 專案，KIP-1034 值得納入升級評估。

參考資料：

• KIP-1034: Dead letter queue in Kafka Streams
• Apache Kafka 4.2.0 Release Announcement
• Kafka Streams Configs: errors.dead.letter.queue.topic.name