java 如何使用 Spring Kafka 的 Acknowledgement.acknowledge() 方法进行手动提交

How to use Spring Kafka's Acknowledgement.acknowledge() method for manual commit

提问by kumarhimanshu449

I am using Spring Kafka first time and I am not able to use Acknowledgement.acknowledge() method for manual commit in my consumer code as mentioned here https://docs.spring.io/spring-kafka/reference/html/_reference.html#committing-offsets. Mine is spring-boot application. If I am not using manual commit process than my code is working fine. But when I use Acknowledgement.acknowledge() for manual commit it shows error related to bean. Also If I am not using manual commit properly please suggest me the right way to do it.

我第一次使用 Spring Kafka，我无法在我的消费者代码中使用 Acknowledgement.acknowledge() 方法进行手动提交，如这里https://docs.spring.io/spring-kafka/reference/html/_reference 所述。 html#committing-offsets。我的是 spring-boot 应用程序。如果我没有使用手动提交过程，那么我的代码就可以正常工作。但是当我使用 Acknowledgement.acknowledge() 进行手动提交时，它显示了与 bean 相关的错误。另外，如果我没有正确使用手动提交，请建议我正确的方法。

Error message:

错误信息：

***************************
APPLICATION FAILED TO START
***************************

Description:

Field ack in Receiver required a bean of type 'org.springframework.kafka.support.Acknowledgment' that could not be found.


Action:

Consider defining a bean of type 'org.springframework.kafka.support.Acknowledgment' in your configuration.

I googled this error I found that I need to add @Component but that is already there in my consumer code.

我在谷歌上搜索了这个错误，我发现我需要添加 @Component 但这已经存在于我的消费者代码中。

My consumer code looks like this: Receiver.java

我的消费者代码如下所示：Receiver.java

import java.util.concurrent.CountDownLatch;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class Receiver {

    @Autowired
    public Acknowledgment ack;

    private CountDownLatch latch = new CountDownLatch(1);

    @KafkaListener(topics = "${kafka.topic.TestTopic}")
    public void receive(ConsumerRecord<?, ?> consumerRecord){
            System.out.println(consumerRecord.value());
            latch.countDown();
            ack.acknowledge();
    }
}

My producer code looks like this: Sender.java

我的生产者代码如下所示：Sender.java

import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class Sender {

    @Autowired
    private KafkaTemplate<String, Map<String, Object>> kafkaTemplate;

    public void send(Map<String, Object> map){
            kafkaTemplate.send("TestTopic", map);

    }

}

EDIT 1:

编辑 1：

My new consumer code looks like this: Receiver.java

我的新消费者代码如下所示：Receiver.java

import java.util.concurrent.CountDownLatch;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class Receiver {

    private CountDownLatch latch = new CountDownLatch(1);

    @KafkaListener(topics = "${kafka.topic.TestTopic}", containerFactory = "kafkaManualAckListenerContainerFactory")
    public void receive(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack){
            System.out.println(consumerRecord.value());
            latch.countDown();
            ack.acknowledge();
    }
}

I changed my configuration class also:

我也改变了我的配置类：

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

@Configuration
@EnableKafka
public class ReceiverConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String consumerGroupId;

    @Bean
    public Map<String, Object> consumerConfigs() throws SendGridException {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
            return props;

    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory(){
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    /*@Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }*/

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaManualAckListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @Bean
    public Receiver receiver() {
        return new Receiver();
    }
}

After adding containerFactory = "kafkaManualAckListenerContainerFactory" to my receive() method I am getting the below error.

将 containerFactory = "kafkaManualAckListenerContainerFactory" 添加到我的接收（）方法后，我收到以下错误。

***************************
APPLICATION FAILED TO START
***************************

Description:

Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration required a bean of type 'org.springframework.kafka.core.ConsumerFactory' that could not be found.
    - Bean method 'kafkaConsumerFactory' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.ConsumerFactory; SearchStrategy: all) found bean 'consumerFactory'


Action:

Consider revisiting the conditions above or defining a bean of type 'org.springframework.kafka.core.ConsumerFactory' in your configuration.

回答by Artem Bilan

You really should follow documentation:

你真的应该遵循文档：

When using manual AckMode, the listener can also be provided with the Acknowledgment; this example also shows how to use a different container factory.

使用手册时AckMode，监听器也可以提供Acknowledgment；这个例子还展示了如何使用不同的容器工厂。

@KafkaListener(id = "baz", topics = "myTopic",
          containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
    ...
    ack.acknowledge();
}

There is really nowhere noted that Acknowledgmentis a bean. So, change your receive()@KafkaListenermethod signature appropriately and remove that @Autowiredfor suspicious Acknowledgmentbean - it just doesn't exists because this object is a part (header) of each received message.

真的没有任何地方注意到这Acknowledgment是一个bean。因此，receive()@KafkaListener适当更改您的方法签名并删除@Autowired可疑 Acknowledgmentbean 的方法签名- 它只是不存在，因为此对象是每个收到消息的一部分（标头）。

回答by Zim8662

For those still looking for a solution to these errors concerning manual acknowledgment, you don't need to specify containerFactory = "kafkaManualAckListenerContainerFactory", instead you can just add:

对于那些仍在寻找有关手动确认的这些错误的解决方案的人，您无需指定 containerFactory = "kafkaManualAckListenerContainerFactory"，而只需添加：

factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

to your receiver config just before you return the factory object.

在您返回工厂对象之前到您的接收器配置。

Then you also need:

那么你还需要：

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

in consumer config props.

在消费者配置道具中。

So in the end your listener method can simply look like:

所以最后你的监听器方法可以简单地看起来像：

@KafkaListener(topics = "${spring.kafka.topic}")
    private void listen(@Payload String payload, Acknowledgment acknowledgment) {
        //Whatever code you want to do with the payload
        acknowledgement.acknowledge(); //or even pass the acknowledgment to a different method and acknowledge even later
    }