我已經實現了一個簡單的 Kafka 死信記錄處理器。使用控制臺生產者生成的記錄時，它可以完美運行。但是我發現我們的 Kafka Streams 應用程序不能保證為接收器主題生成記錄，每生成一條記錄，偏移量就會增加 1。死信處理器背景：我有一個場景，在發布處理它所需的所有數據之前可能會收到記錄。當流應用程序不匹配記錄以進行處理時，它們將移動到死信主題，而不是繼續向下流。當新數據發布時，我們將來自死信主題的最新消息轉儲回流應用程序的源主題，以便使用新數據進行重新處理。死信處理器：在運行應用程序開始時記錄每個分區的結束偏移量結束偏移量標記停止處理給定死信主題的記錄的點，以避免在重新處理的記錄返回死信主題時出現無限循環。應用程序從上次通過消費者組運行產生的最后一個偏移量恢復。應用程序正在使用事務并KafkaProducer#sendOffsetsToTransaction提交最后產生的偏移量。為了跟蹤我的范圍內的所有記錄何時為主題的分區處理，我的服務將其最后產生的從生產者的偏移量與消費者保存的結束偏移量映射進行比較。當我們到達結束偏移量時，消費者通過暫停該分區KafkaConsumer#pause，當所有分區都暫停時（意味著它們達到保存的結束偏移量）然后調用它退出。Kafka 消費者 API狀態：偏移量和消費者位置 Kafka 為分區中的每條記錄維護一個數字偏移量。此偏移量充當該分區內記錄的唯一標識符，并且還表示消費者在分區中的位置。例如，位置 5 的消費者已經消費了偏移量為 0 到 4 的記錄，接下來將接收偏移量為 5 的記錄。Kafka Producer API引用下一個偏移量也總是 +1。將指定偏移量列表發送給消費者組協調器，并將這些偏移量標記為當前事務的一部分。僅當事務成功提交時，這些偏移量才會被視為已提交。提交的偏移量應該是您的應用程序將使用的下一條消息，即 lastProcessedMessageOffset + 1。但是您可以在我的調試器中清楚地看到，單個分區消耗的記錄不是一次遞增 1...我想這可能是一個 Kafka 配置問題，max.message.bytes但沒有一個真正有意義。然后我想也許是因為加入，但沒有看到任何方式會改變制片人的運作方式。不確定它是否相關，但我們所有的 Kafka 應用程序都在使用 Avro 和 Schema Registry...無論生產方法如何，偏移量是否應該始終遞增 1，或者使用 Kafka 流 API 是否可能無法提供與普通生產者消費者客戶端相同的保證？有什么我完全想念的嗎？
2 回答
繁花不似錦
TA貢獻1851條經驗 獲得超4個贊
消息偏移量增加一并不是官方的 API 約定，即使 JavaDocs 表明了這一點（似乎應該更新 JavaDocs）。
如果你不使用事務，你要么得到至少一次語義，要么沒有保證（有些人稱之為最多一次語義）。對于至少一次，記錄可能被寫入兩次，因此，兩個連續消息的偏移量并沒有真正增加一，因為重復寫入“消耗”了兩個偏移量。
如果您使用事務，則事務的每次提交（或中止）都會將提交（或中止）標記寫入主題 - 這些事務標記也“消耗”一個偏移量（這是您觀察到的）。
因此，通常您不應依賴連續的偏移量。您得到的唯一保證是，每個偏移量在分區內都是唯一的。
萬千封印
TA貢獻1891條經驗 獲得超3個贊
我知道知道消息的偏移量可能很有用。但是，Kafka 只會保證 message-X 的偏移量大于最后一條消息（X-1）的偏移量。順便說一句，理想的解決方案不應基于偏移計算。
在后臺，kafka 生產者可能會嘗試重新發送消息。此外，如果經紀人倒閉，則可能會發生重新平衡。恰好一次語義可能會附加一條附加消息。因此，如果發生上述任何事件，您的消息偏移量可能會發生變化。
Kafka 可能會出于內部目的向主題添加其他消息。但是 Kafka 的消費者 API 可能會丟棄這些內部消息。因此，您只能看到您的消息，并且您的消息的偏移量不一定會增加 1。
添加回答
舉報
0/150
提交
取消