Kafka消費異常處理

 

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:775)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:726)

工作中遇到的kafka異常和解決辦法 記錄一下

原文地址https://blog.csdn.net/shibuwodai_/article/details/80678717

異常

異常的主要信息：

a) CommitFailedException

b) Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

其實如果我們對其中的參數，或是對消費的機制比較了解，這個問題就很好解決。當我看到這個異常，我很開心，因為我知道我能通過此異常了解一下Kafka Consumer 消費消息的大致過程。心態是好的~~~

其實現在看這個異常是說：該Consumer不能提交offset了，因為它已經出局了，是因為你的處理小時時間長於你要報告給server的時間。同時還告訴我們怎么處理：要么增加超時時間，要么減少每次poll回來的消息個數。

主要問題在於，何為session timeout？maximum size of batches？poll(timeout)中timeout什么意思？

處理過程

a) 找官網doc

版本：1.1.0

有效信息：

換成通俗易懂的人話：

 

poll() API 主要是判斷consumer是否還活着，只要我們持續調用poll()，消費者就會存活在自己所在的group中，並且持續的消費指定partition的消息。底層是這么做的：消費者向server持續發送心跳，如果一個時間段（session.timeout.ms）consumer掛掉或是不能發送心跳，這個消費者會被認為是掛掉了，這個Partition也會被重新分配給其他consumer

 

下邊這個例子如果理解不上，請通讀全文后，再回來理解一下筆者的意思

------------------------------------分割線------------------------------------

背景：你是個搬磚的，同時還是個瓦工，en….你還有個兒子

上述錯誤就是：工頭命令每個碼農（consumer）最多10分鍾把一個100塊轉運到目的地並把搬來的磚壘房子，然后回來接着取磚、壘房子。問題在於，你搬了100塊磚走了，但是10分鍾過去了，你還沒回來，那我怎么知道你是不是偷懶睡覺去了，工頭就把這個搬磚壘房子的活分給同在一起干活的其他人了（同group不同consumer）。其實你可能沒有偷懶，是因為你太追求完美了（估計是處女座，或是壘自家的房子），壘房子的時間很長（spending too much time message processing），10分鍾內沒能回來向工頭報道，這時，你就得和工頭商量，兩種辦法：1、能不能15分鍾內回來就行，2、10分鍾內回來，但每次搬80塊磚來壘房子。

如果老板是個比較有控制欲的人，對於第二中辦法，同樣的工作量，你無非是多跑幾趟。還能很好的控制你；但是對於第一種辦法，老板是不願意的，為什么，因為和你一起搬磚的還有其他人，他可以協調（rebalance）其他5分鍾就回來的人來干你的活。你告訴他15分鍾對於工頭來說是相對不可控的。

當然你還有兩位一種辦法，你可以找你兒子來搬磚（另起一個線程），你來壘房子，等你壘完了100塊轉，你兒子去告訴工頭，並搬回下一個100塊轉。但是要注意有一個問題，就是你兒子不能在你還沒有壘完上一個100塊轉前就報告給工頭，去獲取下一批100塊轉。這樣你就處理不過來了。

------------------------------------分割線------------------------------------

通過上邊的例子，我們大致清楚了max.poll.interval.ms？maximum size of batches？

max.poll.interval.ms：消費者最大心跳時間間隔

maximum size of batches：消費者每次獲取消息的個數

什么時候發送心跳呢？是poll()方法被調用發送心跳嗎？那poll(timeout)中timeout是什么意思呢？

官網對poll(timeout)中timeout的解釋如下：

1.  
   Parameters:
2.  
   timeout - The time, in milliseconds, spent waiting in poll if data is not available in the buffer. If 0,

   returns immediately with any records that are available currently in the buffer,

   else returns empty. Must not be negative.

這個我費了很大力氣都沒有給它翻譯成人話……