我们正在将我们的kafka实现升级到.9并使用新的消费者java api来创建消费者 . 我正在使用以下代码用于消费者，我们正在使用设置主题到消费者，如在线A和线B是调用我们的服务处理我们收到的消息 . 现在问题是如果我们的消息处理花费超过30秒，我们就会得到Exception .
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("auto.offset.reset", "earliest");
props.put("heartbeat.interval.ms", "1000");
props.put("receive.buffer.bytes", 10485760);
props.put("fetch.message.max.bytes", 5242880);
props.put("enable.auto.commit", false);
//with partition assigned to consumer
KafkaConsumer<Object, Object> consumer = new KafkaConsumer<>(props);
// TopicPartition partition0 = new TopicPartition("TEST-TOPIC", 0);
//consumer.assign(Arrays.asList(partition0));
//assign topic to consumer without partition
//LINE A
consumer.subscribe(Arrays.asList("TEST-TOPIC"), new ConsumerRebalanceListenerImp());
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
try {
ConsumerRecords<Object, Object> records = consumer.poll(1000);
consumeFromQueue(records);//LINE B
consumer.commitSync();
} catch (CommitFailedException e) {
e.printStackTrace();
System.out.println("CommitFailedException");
} catch (Exception e) {
e.printStackTrace();
System.out.println("Exception in while consuming messages");
}
例外是
2016-03-03 10：47：35.095 INFO 6448 --- [ask-scheduler-3] o.a.k.c.c.internals.AbstractCoordinator：标记协调员2147483647死亡 . 2016-03-03 10：47：35.096 ERROR 6448 --- [ask-scheduler-3] oakccinternals.ConsumerCoordinator：在为组TEST-GROUP提交偏移时发生错误ILLEGAL_GENERATION发生组织委员会失败异常org.apache.kafka.clients.consumer.CommitFailedException ：由于org.apache.kafka.clients.consumer.internals.ConsumerCoordinator $ OffsetCommitResponseHandler.handle（ConsumerCoordinator.java:552）中的组重新 balancer 而无法完成提交，位于org.apache.kafka.clients.consumer.internals.ConsumerCoordinator $ OffsetCommitResponseHandler .handle（ConsumerCoordinator.java:493）org.apache.kafka.clients.consumer.internals.AbstractCoordinator $ CoordinatorResponseHandler.onSuccess（AbstractCoordinator.java:665）at org.apache.kafka.clients.consumer.internals.AbstractCoordinator $ CoordinatorResponseHandler位于org.apache.kaf.位于org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient的org.apache.kafka.clients.consumer.internals.RequestFuture.complete（RequestFuture.java:107）的reSuccess（RequestFuture.java:133）$ RequestFutureCompletionHandler.onComplete（ ConsumerNetworkClient.java:380）org.apache.kafka.clients.NetworkClient.poll（NetworkClient.java:274）org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll（ConsumerNetworkClient.java:320）at org .apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll（ConsumerNetworkClient.java:213）
提交偏移时出现上述异常 . 任何建议都会有所帮助，谢谢
2 回答
发生这种情况是因为新的消费者是单线程的，并且它可以通过轮询或提交偏移量来保持心跳的唯一方法是，在30秒后，组协调员将您的消费者标记为已死并呼叫组重新 balancer . 对于这种情况，您可以增加
request.timeout.ms或分割两个线程之间的消耗和处理工作 .您可以通过设置限制poll（）返回的消息数
到某个合适的阈值，该阈值大于您的最大消息，但是如此之低，以至于您每次轮询的消息将减少 .
Kafka 0.10.x支持通过设置明确限制返回给客户端的消息数