Kafka消费者在循环消费消息时丢失消息答案

【问题标题】：Kafka consumer missing messages while consuming messages in loopKafka消费者在循环消费消息时丢失消息
【发布时间】：2021-02-27 15:38:01
【问题描述】：

由于内存限制，我正在循环运行我的消费者代码，提交我的数据然后加载到表中

下面是循环运行的代码

// here is the main part of the component,
// a piece of code executed in the row
// loop
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
System.out.println("Consumer created");
consumer.subscribe(Arrays.asList(topic));
System.out.println("Subscribed to topic " + topic);
try {
    while (pollFlag) {
    ConsumerRecords<String, String> records = consumer.poll(context.consumer_polltime);
     if (records.isEmpty()) {
     globalMap.put("emptyRecordsFlag",false); //Passing the flag value to previous component to end loop
            break;
        }
        for (ConsumerRecord<String, String> record : records) {
            listPayload.add(record.value()); // Adding the messages to list
            i++;
            if(i>=msgbtch)
            {
                pollFlag = false; // Assigning flag value to end the poll at 5000 messages
                break;
            }       
        }
    }
globalMap.put("ConsumerObj",consumer);  
            
}   catch (Exception e) {
            System.out.println("Error Consuming Msg: " + e);
            // TODO: handle exception
            //consumer.close();
    }
row3.payload= String.valueOf(listPayload); // Passing the message data to next component
System.out.println("Committing");
consumer.commitSync();
System.out.println("Closing");
consumer.close();

但由于某种原因，我似乎遗漏了几条消息。我相信这与消费者重新平衡/承诺有关。

如何检查我的消费者是否已准备好从一开始就消费下一批消息而不会丢失任何消息？

【问题讨论】：

是否还有其他具有相同组 id 的消费者？您确定制作人确实发送了您要查找的消息吗？
我自己在主题中推送了 5000 条消息并运行了我的应用程序。但是表格在批处理模式下仅加载了 4917 条消息，而一次加载时加载了 5000 条消息。
再一次，你是如何验证所有 5000 条消息都成功到达主题的？
我在控制中心检查过了。它显示有 5000 条消息滞后。我能够自己弄清楚这个问题。我刚刚删除了以下 if 条件 if(i>=msgbtch) { pollFlag = false; // Assigning flag value to end the poll at 5000 messages break; } 中的中断，甚至在记录中的所有消息都加载到列表之前，上述中断就中断了 for 循环。

标签： java apache-kafka kafka-consumer-api talend

【解决方案1】：

更新： 我能够自己弄清楚这个问题。消息已经下载到记录中，并且在循环时，我已经设置了以下条件

if(i>=msgbtch)
            {
                pollFlag = false; // Assigning flag value to end the poll at 5000 messages
                break;
            }

即使在将所有消息放入列表之前，循环也会中断，并且记录中的所有消息都没有插入列表中。我已经删除了中断条件，它工作正常

【讨论】：