使用新的消费者 api 升级到 .9

Question

我们正在将我们的 kafka 实现升级到 .9 并使用新的消费者 java api 来创建消费者。我正在为消费者使用下面的代码，我们正在使用将主题设置为消费者，如 LINE A 和 LINE 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 错误 6448 --- [ask-scheduler-3] o.a.k.c.c.internals.ConsumerCoordinator：为组 TEST-GROUP 提交偏移量时发生错误 ILLEGAL_GENERATION 提交失败异常 org.apache.kafka.clients.consumer.CommitFailedException：由于组重新平衡，提交无法完成 在 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:552) 在 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:493) 在 org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:665) 在 org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:644) 在 org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:167) 在 org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133) 在 org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107) 在 org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$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) 在 org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)

提交偏移量时出现上述异常。 任何建议都会有所帮助，谢谢

Answer 1

您可以通过设置来限制 poll() 返回的消息数量

max.partition.fetch.bytes

到某个合适的阈值，该阈值大于您的最大消息，但太低以至于您每次投票获得的消息更少。

Kafka 0.10.x 支持通过设置显式限制返回给客户端的消息数量

max.poll.records

Answer 2

发生这种情况是因为新的消费者是单线程的，它可以与消费者组保持心跳的唯一方法是轮询或提交偏移量，30 秒后组协调器将您的消费者标记为死亡并调用进行组再平衡。 对于这种情况，您可以增加request.timeout.ms 或将消费和处理的工作分摊到 2 个线程之间。