我已經(jīng)實現(xiàn)了一個簡單的 Kafka 死信記錄處理器。使用控制臺生產(chǎn)者生成的記錄時，它可以完美運行。但是我發(fā)現(xiàn)我們的 Kafka Streams 應(yīng)用程序不能保證為接收器主題生成記錄，每生成一條記錄，偏移量就會增加 1。死信處理器背景：我有一個場景，在發(fā)布處理它所需的所有數(shù)據(jù)之前可能會收到記錄。當流應(yīng)用程序不匹配記錄以進行處理時，它們將移動到死信主題，而不是繼續(xù)向下流。當新數(shù)據(jù)發(fā)布時，我們將來自死信主題的最新消息轉(zhuǎn)儲回流應(yīng)用程序的源主題，以便使用新數(shù)據(jù)進行重新處理。死信處理器：在運行應(yīng)用程序開始時記錄每個分區(qū)的結(jié)束偏移量結(jié)束偏移量標記停止處理給定死信主題的記錄的點，以避免在重新處理的記錄返回死信主題時出現(xiàn)無限循環(huán)。應(yīng)用程序從上次通過消費者組運行產(chǎn)生的最后一個偏移量恢復(fù)。應(yīng)用程序正在使用事務(wù)并KafkaProducer#sendOffsetsToTransaction提交最后產(chǎn)生的偏移量。為了跟蹤我的范圍內(nèi)的所有記錄何時為主題的分區(qū)處理，我的服務(wù)將其最后產(chǎn)生的從生產(chǎn)者的偏移量與消費者保存的結(jié)束偏移量映射進行比較。當我們到達結(jié)束偏移量時，消費者通過暫停該分區(qū)KafkaConsumer#pause，當所有分區(qū)都暫停時（意味著它們達到保存的結(jié)束偏移量）然后調(diào)用它退出。Kafka 消費者 API狀態(tài)：偏移量和消費者位置 Kafka 為分區(qū)中的每條記錄維護一個數(shù)字偏移量。此偏移量充當該分區(qū)內(nèi)記錄的唯一標識符，并且還表示消費者在分區(qū)中的位置。例如，位置 5 的消費者已經(jīng)消費了偏移量為 0 到 4 的記錄，接下來將接收偏移量為 5 的記錄。Kafka Producer API引用下一個偏移量也總是 +1。將指定偏移量列表發(fā)送給消費者組協(xié)調(diào)器，并將這些偏移量標記為當前事務(wù)的一部分。僅當事務(wù)成功提交時，這些偏移量才會被視為已提交。提交的偏移量應(yīng)該是您的應(yīng)用程序?qū)⑹褂玫南乱粭l消息，即 lastProcessedMessageOffset + 1。但是您可以在我的調(diào)試器中清楚地看到，單個分區(qū)消耗的記錄不是一次遞增 1...我想這可能是一個 Kafka 配置問題，max.message.bytes但沒有一個真正有意義。然后我想也許是因為加入，但沒有看到任何方式會改變制片人的運作方式。不確定它是否相關(guān)，但我們所有的 Kafka 應(yīng)用程序都在使用 Avro 和 Schema Registry...無論生產(chǎn)方法如何，偏移量是否應(yīng)該始終遞增 1，或者使用 Kafka 流 API 是否可能無法提供與普通生產(chǎn)者消費者客戶端相同的保證？有什么我完全想念的嗎？