如何增加Kafka消费者的请求时间

Question

我正在使用Kafka和Spring Listener。下面是一段代码。过去，我们发布了超过10万条消息来测试主题，系统似乎运行良好。但几天前，我更改了消费者的groupId。之后，这个新消费者尝试从一开始就处理所有消息，这需要花费大量时间。但过了一段时间（10秒），经纪人会启动消费者。所以结果没有kafka寄存器来侦听消息。

 @KafkaListener(
            topicPattern = "test",
            groupId = "test",
            id = "test",
            containerFactory = "testKafkaListenerContainerFactory")
    public void consume(@Payload String payload) throws IOException {
    }

Kafka消费者配置：

Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    props.put("security.protocol", "SSL");

然后，我使用cli通过以下命令读取消息并观察到相同的行为。10秒后，消费者停止阅读来自Kafka的消息。

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

如何增加kafka客户端的请求超时时间或其他更好的方法来解决这个问题？

Answer 1

In the past we have published more than 1 lac message to test topic and system seems to be working fine.

你的Kafka有所需的数据吗？Kafka不会永远存储消息。消息的持续时间由代理中的保留设置控制。如果您的数据早于保留期，它将丢失。