Kafka消费者不接收旧消息

Question

Kafka消费者不接收在消费者开始之前产生的消息。

 public class MyKafkaConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final String TOPIC="javaapp";
    private final String BOOTSTRAP_SERVERS="localhost:9092";
    private int receivedCounter=0;
    private ExecutorService executorService=Executors.newFixedThreadPool(1);

    private BlockingQueue<ConsumerRecords<String, String>> queue=new LinkedBlockingQueue<>(500000);
    private MyKafkaConsumer() {
        final Properties props=new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaGroup6");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumer=new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
    }

    public static void main(String[] args) throws InterruptedException {
        MyKafkaConsumer perfKafkaConsumer=new MyKafkaConsumer();
        perfKafkaConsumer.consumeMessage();
        perfKafkaConsumer.runConsumer();
    }

    private void runConsumer() throws InterruptedException {
        consumer.poll(Duration.ofMillis(1000));
        while (true) {
            final ConsumerRecords<String, String> consumerRecords=consumer.poll(Duration.ofMillis(10000));
            if (!consumerRecords.isEmpty()) {
                System.out.println("Adding result in queue " + queue.size());
                queue.put(consumerRecords);
            }
            consumer.commitAsync();

        }
    }

    private void consumeMessage() {
        System.out.println("Consumer starts at " + Instant.now());
        executorService.submit(() -> {
            while (true) {
                ConsumerRecords<String, String> poll=queue.take();
                poll.forEach(record -> {
                    System.out.println("Received " + ++receivedCounter + " time " + Instant.now(Clock.systemUTC()));
                });

            }
        });
    }
}

ConsumerRecords始终为空

虽然，如果我启动我的消费者比生产者比它接收消息。（Kafka-客户端版本2.4.1)

Answer 1

AUTO.OFFSET.RESET使用者设置控制新使用者组将从何处开始使用某个主题。默认情况下，它被设置为“最新”，这会将消费者组的偏移量设置为最新的偏移量。如果所有消费者组都应从主题中最早的偏移量开始，则您希望将此设置为“最早”。