unable to to poll the message from the kafka server by using the consumer

Viewed 12

while I run the java code use the kafka comsumer to poll the message from the kafka server, it always show 0 message,

the kafka server log shows the group error:

[2022-09-08 05:03:03,785] INFO [GroupCoordinator 1]: Dynamic Member with unknown member id joins group KafkaStudy in Empty state. Created a new member id consumer-1-e99d5a30-346b-4e26-8ff5-7f0e5fac3e9b for this member and add to the group. (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:03,785] INFO [GroupCoordinator 1]: Preparing to rebalance group KafkaStudy in state PreparingRebalance with old generation 27 (__consumer_offsets-15) (reason: Adding new member consumer-1-e99d5a30-346b-4e26-8ff5-7f0e5fac3e9b with group instance id None; client reason: not provided) (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:03,788] INFO [GroupCoordinator 1]: Stabilized group KafkaStudy generation 28 (__consumer_offsets-15) with 1 members (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:03,809] INFO [GroupCoordinator 1]: Assignment received from leader consumer-1-e99d5a30-346b-4e26-8ff5-7f0e5fac3e9b for group KafkaStudy for generation 28. The group has 1 members, 0 of which are static. (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:13,819] INFO [GroupCoordinator 1]: Member consumer-1-e99d5a30-346b-4e26-8ff5-7f0e5fac3e9b in group KafkaStudy has failed, removing it from the group (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:13,820] INFO [GroupCoordinator 1]: Preparing to rebalance group KafkaStudy in state PreparingRebalance with old generation 28 (__consumer_offsets-15) (reason: removing member consumer-1-e99d5a30-346b-4e26-8ff5-7f0e5fac3e9b on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator)

[2022-09-08 05:03:13,821] INFO [GroupCoordinator 1]: Group KafkaStudy with generation 29 is now empty (__consumer_offsets-15) (kafka.coordinator.group.GroupCoordinator)

the below are code:

 Properties properties = new Properties();
        properties.put("bootstrap.servers","192.168.226.133:9092");
        properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("enable.auto.commit", true);
        properties.put("group.id","KafkaStudy");
        properties.put("group.instance.id","1");
        consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Collections.singleton("quickstart-events"));

        try{
               ConsumerRecords<String,String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {System.out.println(record.value());}

            } finally{
            consumer.close();
        }
1 Answers

By default, the consumer starts at the end of the topic. You need a new group id with

properties.put("auto.offset.reset","earliest");

Or produce data while consumer is running.

Related