kafka消费问题处理记录

r囧r小猫 2022-04-12 13:48 219阅读 0赞

公司系统内部数据交换使用了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数据消费的正常过程；  
①消费者订阅专题→②生产者生成数据→③消费者收到数据→④消费者处理数据→⑤消费者提交偏移量；

正常的消费当然不会有问题，但想象下，如果④这一步处理太慢，很长时间还没有执行到⑤这一步骤，或者是消费者出问题了直接崩溃了， 这个时候服务器应该如何处理？

如果它不管分配给该消费者的这批数据，那么这批数据就永久的丢失了。这种情况很明显无法容忍。  
kafka通过引入一个超时时间来处理这种情况，[这个超时时间参数就是max.poll.interval.ms][max.poll.interval.ms]，它表示的是从③→⑤这一段时间，如果超过了超时时间，那么服务器认为这个消费者有问题，因此它会将原本分配给该消费者的数据重新进行分配。如果之后消费者再提交偏移量，服务端会返回一个错误给消费者，这就是上面看到的错误的原因。

如果服务器每次给消费者分配一条记录，那么消费者与服务器之间的交互会非常地频繁，在网络上的开销也会非常大。一般情况下消费者都是批量消费的，来减少网络上的开销。每次最多消费多少条记录通过参数max.poll.records来进行控制。

再来看之前的消费过程，什么情况下消费者会超时导致数据重复分配？根据上面的分析，如果在超时时间到消费者还未处理完时就会导致重新分配。倘若超时时间设置的很短，或者一次最多处理的记录数非常大，就很有可能导致该问题。

综上，在出现这种情况时，可以进行如下处理：  
一是搞清楚消费者处理慢到底是在哪一过程，对这一过程进行集中的优化，如多线程处理、使用Redis替代MySql等传统数据库等；  
二是优化完成后如果还有问题，可以调整超时时间及每次最大处理记录数两个参数，调大超时时间或者调小每次最大处理记录数，以此来降低超时的机率。

回到公司的问题，将两个参数写到了配置文件中（之前是直接写死的），然后定义了一个对象来使用ConfigurationProperties加载配置项。  
初始化消费者代码如下：

property = new Properties();
    	property.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,serverNodes);
    	property.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    	//property.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    	property.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    	property.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);
    	property.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    	property.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
    	property.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 20000);
    	property.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
    	property.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 512 * 1024);
    
    	property.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaProperties.getMaxPollIntervalMs());
    	property.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaProperties.getMaxPollRecords());
    	property.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 40000);
    
    	consumer = new KafkaConsumer<>(property);

改造完后发现消费者收不到消息！  
通过调试最终发现，在new KafkaConsumer<>(property)时一直未返回。  
一步步调试下去，最终发现在ConfigDef类的parseType方法中，如果发现属性参数与预期的不一致时，会抛出异常，如下：

case INT:
    	if (value instanceof Integer) { 
    		return value;
    	} else if (value instanceof String) { 
    		return Integer.parseInt(trimmed);
    	} else { 
    		throw new ConfigException(name, value, "Expected value to be a 32-bit integer, but it was a " + value.getClass().getName());
    	}

问题很明显了，在配置类中属性类型是long，而kafka对要求这两个参数的类型都是Integer，因此会抛出异常。  
可见，在使用ConfigurationProperties来加载kafka的配置项时，一定要注意参数类型与预期类型匹配。每一个参数类型在ConsumerConfig中可以查到。

[max.poll.interval.ms]: http://xn--max-i68du9wetnnxpu4aa78cm575ak3fx8v.poll.interval.ms