Apache的卡夫卡消费者停止和开始之间丢失的邮件(messages lost between Ap

我是新来卡夫卡和使用Apache卡夫卡消费者阅读生产的消息。 但是，当我停止并启动一定的时间。 所有产生的消息之间都将丢失。 如何处理这种情况。 我使用这些属性“auto.offset.reset”，“最新”和“enable.auto.commit”，“假”。

这是我using.Any帮助表示赞赏代码。

Properties props = new Properties();
        props.put("bootstrap.servers", localhost:9092);
        props.put("group.id", "service");
        props.put("enable.auto.commit", "false"); 
        props.put("auto.offset.reset", "latest"); 
        props.put("key.deserializer", KAFKA_DESERIALIER_STRING_KEYVALUE);
        props.put("value.deserializer", KAFKA_DESERIALIER_STRING_KEYVALUE);

        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topicname));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {        
                JSONObject jsonObj = new JSONObject(record.value());
                JdbcUtilToUdm.insertdataintodb(args, jsonObj);   
            }
        }   

Answer 1:

你必须显式调用consumer.commitSync（）或consumer.commitAsync（），因为禁用了自动提交。 你可以同步或不取决于你所需要/喜欢哪种方式提交。 这是如何在日志中的消费群位置将被保留。 你应该通话记录进行处理后提交（你已经完成了所有刀片所以大概之后，但在你的情况再次轮询之前）。

文章来源: messages lost between Apache kafka consumer stop and start