kafka-生产环境问题-报错Maximum application poll interval

Posted 2023-03-31 健康-是最好的时光

一.产生的问题

在.NET环境下使用kafka，消费者长时间消费，会报“ Application maximum poll interval (10000ms) ”错误。

二.重现问题

2.1.消费者配置

SessionTimeoutMs（会话超时时间）和MaxPollIntervalMs（上一次拉取消息和本次拉取消息之间的时间间隔）配置为10秒。

 var config = new ConsumerConfig
 
     BootstrapServers = brokerServer,
     //同组轮询，不同组广播
     GroupId = groupName,
 
     EnableAutoCommit = false,
 
     AutoOffsetReset = AutoOffsetReset.Earliest,
 
     EnablePartitionEof = true,
     PartitionAssignmentStrategy = PartitionAssignmentStrategy.Range,
       
     //心跳(<）超时
     SessionTimeoutMs = 10000,
     //上一次拉取消息和本次拉取消息之间的时间间隔
     MaxPollIntervalMs = 10000,
 ;

2.2.模拟一个耗时的业务

DoBussinessForLongTime()方法中特意休眠12秒，模拟做业务慢场景。

 ConsumeResult<Ignore, string> consumeResult = null;
 
 while (true)
 
     try
     
         //拉取kafka消息
         consumeResult = consumer.Consume(cancellationToken);
         #region 模拟长时间做公司业务
         
         　　DoBussinessForLongTime(consumeResult);
         
         #endregion
 
         consumer.Commit(consumeResult);
     
     catch (ConsumeException e)
     
         Console.WriteLine($"Consume error: e.Error.Reason");
     
     catch (KafkaException e)
     
         Console.WriteLine($"Commit error: e.Error.Reason");
     
     catch (OperationCanceledException e)
     
         throw e;
     
     catch (Exception e)
     
         Console.WriteLine($"Commit error: e");
     
 
 
 
 /// <summary>
 /// 长时间做公司业务
 /// 配合SessionTimeoutMs = 10000，MaxPollIntervalMs = 10000使用
 /// </summary>
 /// <param name="consumeResult"></param>
 private static void DoBussinessForLongTime(ConsumeResult<Ignore, string> consumeResult)
 
     Console.WriteLine($"DoBussinessForLongTime Start DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff")");
     Thread.Sleep(12000);
     Console.WriteLine($"DoBussinessForLongTime End DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff")");
 

给名为MyTopic的主题生产3条消息，分别为a、b、c。

我们最终看到在拉取第2条消息时，报ConsumeException异常了，红色框内的异常是我们主动打印出来的，黄色框内的异常是.NET自己打印出来的。我们来翻译下这句话“ Application maximum poll interval (10000ms) exceeded by 183ms (adjust max.poll.interval.ms for long-running message processing): leaving group”，这句话的意思是说：拉取最大时间间隔为10秒，超过了183毫秒，离开组。

详细解释下这个报错是什么意思：这个报错是指你上一次拉取消息和本次拉取消息之间的时间间隔超过了kafka允许的最大的拉取消息的时间间隔（kafka里默认配置是5分钟，我们修改了消费者配置是10秒），也就是说kafka认为你拉取消息后做了大于10秒的业务，拉到消息是要去做业务的（每家公司做的业务不同），什么业务能做10秒呢？（做完业务才再去拉取下一条消息），所以kafka的组协调器（GroupCoordinator）就认为你这个消费者（每一个消费者都归属于一个消费者组）不给力，觉的你干活太慢了，做业务太慢了，把你这个消费者从消费者组移除出去，就报错了。

这个问题产生的原因：是咱们引入的第三方的dll（.NET里引用的是Confluent.Kafka.dll）在长时间消费过程中，会偶发判断上一次拉取消息和本次拉取消息之间的时间间隔有误（怀疑），明明没有超过5分钟（kafka里默认配置是5分钟），但是确判断是超过了5分钟。

 

三.如何解决这个问题？

使用异常过滤器，捕获到“Application maximum poll interval”异常，记录下日志就好了，然后继续去拉取本次失败的消息（前提是已经对之前拉取到的消息做了偏移量提交）。

  catch (ConsumeException e) when (e.Error.Reason.Contains("Application maximum poll interval"))//满足条件才进入catch
  
  　　//此处记录下日志
  

 

旧书不厌百回读，熟读深思子自知。