Блог

У меня есть следующий код для подключения к Kafka

 Properties props = new Properties();
props.put("bootstrap.servers", "myconfluentkafkabroker:9092");
props.put("group.id","test");
props.put("enable.auto.commit","true");
props.put("auto.commit.interval.ms","1000");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_CG");
props.put("group.instance.id", "my_instance_CG_id");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put("key.deserializer", Class.forName("org.apache.kafka.common.serialization.StringDeserializer"));
props.put("value.deserializer", Class.forName("org.apache.kafka.common.serialization.StringDeserializer"));

KafkaConsumer<String,String> consumer = new KafkaConsumer<String,String>(props);
consumer.subscribe(Arrays.asList("MyTopic"));
try {
    while (true) { 
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records)
        {
            log.debug("topic = %s, partition = %d, offset = %d,"
                customer = %s, country = %sn",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value());

            int updatedCount = 1;
            if (custCountryMap.countainsKey(record.value())) {
                updatedCount = custCountryMap.get(record.value())   1;
            }
            custCountryMap.put(record.value(), updatedCount)

            JSONObject json = new JSONObject(custCountryMap);
            System.out.println(json.toString(4));
        }
    }
} finally {
    consumer.close();
}
 

Код не выдал никаких ошибок, но я все еще не вижу в списке потребителя

будет ли это проблемой?

props.put(«group.instance.id «, «my_instance_CG_id»);

Вы должны проверить информацию, которую вы видите, с помощью встроенных инструментов, которые предоставляет Kafka, например kafka-consumer-groups.sh

Вам также потребуется фактически опрашивать сообщения и фиксировать смещения, а не просто подписываться, прежде чем вы что-либо увидите.

В противном случае для этой конкретной панели мониторинга центра управления может потребоваться добавить перехватчики мониторинга в ваш клиент