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();
}