Потребитель Кафки не возвращает никаких записей - PullRequest
Новая
       5

Потребитель Кафки не возвращает никаких записей

0 голосов
/ 10 декабря 2018

Я пытаюсь сделать маленький PoC с Кафкой.Однако при создании потребителя в Java этот потребитель не получает сообщений.Даже когда я запускаю файл kafka-console-consumer.sh с тем же URL-адресом / темой, я получаю сообщения.Кто-нибудь знает, что я могу сделать не так?Этот код вызывается GET API. 

public List<KafkaTextMessage> receiveMessages() {
    log.info("Retrieving messages from kafka");
    val props = new Properties();
    // See https://kafka.apache.org/documentation/#consumerconfigs
    props.put("bootstrap.servers", "my-cluster-kafka-bootstrap:9092");
    //props.put("client.id", "my-topic consumer");
    props.put("group.id", "test");
    props.put("enable.auto.commit", "false");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    ImmutableList.Builder<KafkaTextMessage> builder;
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        consumer.subscribe(Collections.singletonList(TEXT_MESSAGE_TOPIC));
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        builder = ImmutableList.builder();
        for (ConsumerRecord<String, String> record : records) {
            builder.add(new KafkaTextMessage(record.value()));
            log.info("We got at position: {} key:{} value: {}", record.offset(), record.key(), record.value());
            consumer.commitSync();
        }
    }
    return builder.build();
}

1 Ответ

0 голосов
/ 10 декабря 2018

Попробуйте добавить auto.offset.reset=earliest в ваших потребительских свойствах.По умолчанию установлено значение latest.Я предлагаю это, потому что вижу, что ваше group.id установлено на test, значение, которое вы, возможно, уже использовали в предыдущих тестах.

...