模拟Kafka CommitFailedException

Question

我试图模拟Kafka抛出的CommitFailedException。

我手动将“session.timeout.ms”设置为10000毫秒，将“enable.auto.commit”设置为false。

在kafkaconsumer.poll（）之后，我有一个语句thread.sleep（12000），然后我执行提交。我预计，由于线程在下一轮轮询之前的时间为12s，所以使用者应该被标记为已死，并且应该抛出一个CommitFailedException。然而，该过程执行得很顺利。

如何模拟KafkaConsumer引发的异常。

consumer.subscribe(Arrays.asList("foo"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
            }

            try {
                Thread.sleep(12000);
            }catch (Exception e){
                e.printStackTrace();
            }
            consumer.commitSync();
        }

Answer 1

Kafka通过一个单独的线程使用心跳机制来检查消费者的健康状况。使用者心跳线程必须在session.timeout.ms时间到期之前向代理发送心跳。

heartbeat.interval.ms：在使用Kafka的组管理设施时，向消费者协调器发送心跳的预期间隔时间。heartbeat用于确保消费者的会话保持活动，并在新消费者加入或离开组时促进重新平衡。

session.timeout.ms：当使用Kafka的组管理工具时，用于检测客户端故障的超时。客户端向代理发送周期性心跳以指示其活跃度。如果在此会话超时到期之前，代理没有接收到心跳，则代理将从组中删除此客户端并启动重新平衡。

max.poll.interval.ms：使用使用者组管理时，调用poll（）之间的最大延迟。这为使用者在获取更多记录之前可以空闲的时间量设置了一个上限。如果在超时到期之前未调用poll()，则认为使用者失败，组将重新平衡以便将分区重新分配给另一个成员。

如果使用者由于session.timeout.ms中没有心跳或max.poll.interval.ms中没有轮询而被Kafka认为已死亡，使用者将无法提交消息并获得CommitFailedException。

CommitFailedException：当使用KafKaConsumer.CommitSync（）提交的偏移量失败并出现不可恢复的错误时，会引发此异常。当组重新平衡在成功应用提交之前完成时，可能会发生这种情况。在这种情况下，通常不能重试提交，因为某些分区可能已经分配给组中的另一个成员。

结果；因为heartbeat线程是一个单独的线程，所以代码中的睡眠不会影响它。但在您的情况下，可以将max.poll.interval.ms设置为10秒，以获得CommitFailedException。