Apache Kafka 中的消息消费确认答案

【问题标题】：Message consumption acknowledgement in Apache KafkaApache Kafka 中的消息消费确认
【发布时间】：2018-06-06 23:53:19
【问题描述】：

我实现了一个 Java Consumer，它使用来自 Kafka 主题的消息，然后将这些消息与 POST 请求一起发送到 REST API。

    while (true) {
        ConsumerRecords<String, Object> records = consumer.poll(200);
        for (ConsumerRecord<String, Object> record : records) {
            CloseableHttpClient httpClient = HttpClientBuilder.create().build();  
            Object message = record.value();
            JSONObject jsonObj = new JSONObject(message.toString());
            try {
                HttpPost request = new HttpPost(this.getConsumerProperties().getProperty("api.url"));
                StringEntity params = new StringEntity(message.toString());
                request.addHeader("content-type", "application/json");
                request.addHeader("Accept", "application/json");
                request.setEntity(params);

                CloseableHttpResponse response = httpClient.execute(request);
                HttpEntity entity = response.getEntity();
                String responseString = EntityUtils.toString(entity, "UTF-8");
                System.out.println(message.toString());
                System.out.println(responseString);
            } catch(Exception ex) {
              ex.printStackTrace();
            }  finally {
                try {   
                    httpClient.close(); 
                } catch(IOException ex) {
                    ex.printStackTrace();
                } 
            }                  
        } 
    }

假设一条消息已被消费，但 Java 类未能访问 REST API。该消息将永远不会被传递，但它会被标记为已使用。处理此类案件的最佳方法是什么？当且仅当来自 REST API 的响应成功时，我能否以某种方式确认消息？

【问题讨论】：

标签： java apache-kafka

【解决方案1】：

在消费者属性中，将 enable.auto.commit 设置为 false。这意味着提交补偿的责任在于消费者。

因此，在上面的示例中，您可以根据 response.statusCode 选择通过调用 consumer.commitAsync() 来提交偏移量。

【讨论】：