今天在尝试自己写一个消费者时,遇到了如下问题:
消费者代码基本和官网一样如下:
Properties props = new Properties();
props.put("bootstrap.servers", "host-129-152:9092");
props.put("group.id", "test1");
props.put("client.id", "test1");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset","earliest");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
try{
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic11"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
}catch(Exception e){
logger.error(e.getMessage());
}
运行过程中无法消费消息,并报如下错误:
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator]Received GroupCoordinator response ClientResponse(receivedTimeMs=1525675045460, latencyMs=3, disconnected=false, requestHeader={api_key=10,api_version=1,correlation_id=12,client_id=test1}, responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1 rack: null))) for group test1
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator]Group coordinator lookup for group test1 failed: The coordinator is not available.
[org.apache.kafka.clients.consumer.internals.AbstractCoordinator]Coordinator discovery failed for group test1, refreshing metadata
请问可能是什么原因造成这个错误的呢?我在客户端命令可以正常消费信息,所用kakfa版本是0.11