KafkaConsumer не использует сообщения

Из того, что я прочитал до сих пор, потребитель должен правильно потреблять сообщения, но единственный вывод, который я получаю, - это «.».

public class KafkaConsumerSample {

public static void main(String []args) {
    runProducer();
    //runConsumer();
}

static void runProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    Producer<Long, String> producer = new KafkaProducer<Long, String>(props);

     for (int i = 0; i < 10; i++) {
         producer.send(new ProducerRecord<Long, String>("foo", "test message" + Integer.toString(i)));
     }
     producer.close();
}

static void runConsumer() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props); 
    consumer.subscribe(Arrays.asList("foo"));

    while(true) {
        ConsumerRecords<Long, String> records = consumer.poll(Duration.ofSeconds(10));
        System.out.println(".");
        for(ConsumerRecord<Long, String> record: records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

}

Журнал Kafka сообщает мне следующее:

Member consumer-1 in groupp test has failed, removing it from the group

Preparing to rebalance group test with old generation 21

Group test with generation 22 is now empty

Я не очень понимаю, что это значит. Что мне не хватает? Мне это кажется очень простым.

Обновить

Я хотел бы прочитать все смещения один раз, когда я запускаю потребителя. От смещения 0 до последнего смещения. Я пытался :

static void runConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OffsetResetStrategy.EARLIEST.toString().toLowerCase());
        KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props); 
        consumer.subscribe(Arrays.asList("foo"));


        while(true) {
            consumer.seekToBeginning(consumer.assignment());
            ConsumerRecords<Long, String> records = consumer.poll(Duration.ofSeconds(5));

            System.out.println(".");
            for(ConsumerRecord<Long, String> record: records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }

Просто установить AUTO_OFFSET_RESET_CONFIG на самое раннее, по-видимому, недостаточно. Я запустил потребитель один раз, и он напечатал все смещения от 0 до последнего, но теперь он больше не работает после повторного запуска? Не уверен, что правильно понимаю этот конфиг. Вот почему я добавил метод Consumer.seek(), но где именно его разместить? Если я помещу его в цикл while(true), как показано, он печатает все от смещения 0 до последнего, но не каждые 5 секунд, а без перерыва... постоянно спамит мою консоль.