I have the following structure:
zookeeper: 3.4.12
kafka: kafka_2.11-1.1.0
server1: zookeeper + kafka
server2: zookeeper + kafka
server3: zookeeper + kafka
Created topic with replication factor 3 and partitions 3 by kafka-topics shell script.
./kafka-topics.sh --create --zookeeper localhost:2181 --topic test-flow --partitions 3 --replication-factor 3
And use group localConsumers. it works fine when leader is ok.
./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-flow
Topic:test-flow PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test-flow Partition: 0 Leader: 3 Replicas: 3,2,1 Isr: 3,2,1
Topic: test-flow Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,3,2
Topic: test-flow Partition: 2 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3
Consumers’ log
Received FindCoordinator response ClientResponse(receivedTimeMs=1529508772673, latencyMs=217, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1, clientId=consumer-1, correlationId=0), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=NONE, node=myserver3:9092 (id: 3 rack: null)))
But if leader is down — I get the error in consumer (systemctl stop kafka):
Node 3 is unavailable. ok
./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-flow
Topic:test-flow PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test-flow Partition: 0 Leader: 2 Replicas: 3,2,1 Isr: 2,1
Topic: test-flow Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,2
Topic: test-flow Partition: 2 Leader: 2 Replicas: 2,1,3 Isr: 2,1
Consumers’ log
Received FindCoordinator response
ClientResponse(receivedTimeMs=1529507314193, latencyMs=36,
disconnected=false,
requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1,
clientId=consumer-1, correlationId=149),
responseBody=FindCoordinatorResponse(throttleTimeMs=0,
errorMessage='null', error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1
rack: null)))
- Group coordinator lookup failed: The coordinator is not available.
- Coordinator discovery failed, refreshing metadata
Consumer unable to connect until leader is down or reconnect with another consumer group.
Can’t understand why it happens?
Consumer should be rebalanced to another broker, but it doesn’t.
Hi. I have the latest version of kafka-python.
from kafka import KafkaConsumer consumer = KafkaConsumer('site_index_queue', group_id='test_group', bootstrap_servers=['kafka201:9092','kafka202:9092','kafka203:9092']) for message in consumer: print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value))
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 394
DEBUG:kafka.conn: Request 5: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.client:Initializing connection to node 392 for metadata request
DEBUG:kafka.client:Initiating connection to node 392 at kafka203:9092
DEBUG:kafka.conn:: creating new socket
DEBUG:kafka.client:Initializing connection to node 393 for metadata request
DEBUG:kafka.client:Initiating connection to node 393 at kafka201:9092
DEBUG:kafka.conn:: creating new socket
DEBUG:kafka.conn:: established TCP connection
DEBUG:kafka.client:Node 392 connected
DEBUG:kafka.conn:: established TCP connection
DEBUG:kafka.client:Node 393 connected
DEBUG:kafka.client:Sending metadata request MetadataRequest_v0(topics=[‘site_index_queue’]) to node 393
DEBUG:kafka.conn: Request 1: MetadataRequest_v0(topics=[‘site_index_queue’])
DEBUG:kafka.conn: Response 5: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 392
DEBUG:kafka.conn: Request 1: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.conn: Response 1: MetadataResponse_v0(brokers=[(node_id=392, host=u’kafka203′, port=9092), (node_id=393, host=u’kafka201′, port=9092), (node_id=394, host=u’kafka202′, port=9092)], topics=[(error_code=0, topic=u’site_index_queue’, partitions=[(error_code=0, partition=2, leader=392, replicas=[392, 393, 394], isr=[392, 394, 393]), (error_code=0, partition=1, leader=394, replicas=[394, 392, 393], isr=[392, 394, 393]), (error_code=0, partition=0, leader=393, replicas=[393, 394, 392], isr=[392, 394, 393])])])
DEBUG:kafka.cluster:Updated cluster metadata to Cluster(brokers: 3, topics: 1, groups: 0)
DEBUG:kafka.conn: Response 1: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 394
DEBUG:kafka.conn: Request 6: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.conn: Response 6: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 394
DEBUG:kafka.conn: Request 7: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.conn: Response 7: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 394
DEBUG:kafka.conn:<BrokerConnection host=kafka202/144.76.183.202 port=9092> Request 8: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.conn: Response 8: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
DEBUG:kafka.coordinator:Sending group coordinator request for group test_group to broker 393
DEBUG:kafka.conn: Request 2: GroupCoordinatorRequest_v0(consumer_group=’test_group’)
DEBUG:kafka.conn: Response 2: GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)
DEBUG:kafka.coordinator:Group Coordinator Not Available; retry
What’s wrong? What does this (DEBUG:kafka.coordinator:Received group coordinator response GroupCoordinatorResponse_v0(error_code=15, coordinator_id=-1, host=u», port=-1)) mean? How can I fix this?
Thanks for any help.
Когда я пишу тему в kafka, возникает ошибка: Offset commit failed:
2016-10-29 14:52:56.387 INFO [nioEventLoopGroup-3-1][org.apache.kafka.common.utils.AppInfoParser$AppInfo:82] - Kafka version : 0.9.0.1
2016-10-29 14:52:56.387 INFO [nioEventLoopGroup-3-1][org.apache.kafka.common.utils.AppInfoParser$AppInfo:83] - Kafka commitId : 23c69d62a0cabf06
2016-10-29 14:52:56.409 ERROR [nioEventLoopGroup-3-1][org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$DefaultOffsetCommitCallback:489] - Offset commit failed.
org.apache.kafka.common.errors.GroupCoordinatorNotAvailableException: The group coordinator is not available.
2016-10-29 14:52:56.519 WARN [kafka-producer-network-thread | producer-1][org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater:582] - Error while fetching metadata with correlation id 0 : {0085000=LEADER_NOT_AVAILABLE}
2016-10-29 14:52:56.612 WARN [pool-6-thread-1][org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater:582] - Error while fetching metadata with correlation id 1 : {0085000=LEADER_NOT_AVAILABLE}
При создании новой темы с помощью команды это нормально.
./kafka-topics.sh --zookeeper localhost:2181 --create --topic test1 --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1
Это код производителя, использующий Java ：
public void create() {
Properties props = new Properties();
props.clear();
String producerServer = PropertyReadHelper.properties.getProperty("kafka.producer.bootstrap.servers");
String zookeeperConnect = PropertyReadHelper.properties.getProperty("kafka.producer.zookeeper.connect");
String metaBrokerList = PropertyReadHelper.properties.getProperty("kafka.metadata.broker.list");
props.put("bootstrap.servers", producerServer);
props.put("zookeeper.connect", zookeeperConnect);//声明ZooKeeper
props.put("metadata.broker.list", metaBrokerList);//声明kafka broker
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 1000);
props.put("linger.ms", 10000);
props.put("buffer.memory", 10000);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<String, String>(props);
}
Где не так？
5 ответов
Лучший ответ
Я столкнулся с подобной проблемой. Проблема, с которой я столкнулся, заключалась в том, что когда вы запускаете своего брокера kafka, с ним связано свойство «KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR». Если вы работаете с кластером с одним узлом, убедитесь, что для этого свойства установлено значение «1». По умолчанию установлено значение 3. Это изменение решило мою проблему. (вы можете проверить значение в файле kafka.properties) Примечание: я использовал базовый образ confluent kafka версии 4.0.0 (confluentinc / cp-kafka: 4.0.0)
14
Vishal Akkalkote
4 Июн 2020 в 09:19
Глядя на ваши журналы, проблема в том, что у кластера, вероятно, нет подключения к узлу, который является единственной известной репликой данной темы в zookeeper.
Вы можете проверить это с помощью данной команды:
kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1
Или используя kafkacat:
kafkacat -L -b localhost:9092
Пример результата:
Metadata for all topics (from broker 1003: localhost:9092/1003):
1 brokers:
broker 1003 at localhost:9092
1 topics:
topic "topic1" with 1 partitions:
partition 0, leader -1, replicas: 1001, isrs: , Broker: Leader not available
Если у вас кластер с одним узлом, идентификатор брокера ( 1001 ) должен совпадать с лидером раздела topic1.
Но, как вы можете видеть, единственной известной репликой topic1 была 1001 , которая сейчас недоступна, поэтому нет возможности воссоздать тему на другом узле.
Источником проблемы может быть автоматическое создание идентификатора брокера (если вы не указали broker.id или для него установлено значение -1).
Затем при запуске брокера (тот же единственный брокер) вы, вероятно, получите идентификатор брокера, отличный от того, который был ранее и отличный от того, который был отмечен в zookeeper (это причина, по которой удаление раздела может помочь — но это не производственное решение ).
Решением может быть установка значения broker.id в конфигурации узла на фиксированное значение — согласно документации это должно быть сделано в производственной среде:
broker.id=1
Если все в порядке, вы должны получить что-то вроде этого:
Metadata for all topics (from broker 1: localhost:9092/1001):
1 brokers:
broker 1 at localhost:9092
1 topics:
topic "topic1" with 1 partitions:
partition 0, leader 1, replicas: 1, isrs: 1
Документация Kafka: https://kafka.apache.org/documentation/#prodconfig
3
tk3
18 Май 2018 в 13:02
Привет, вы должны сохранить свои реплики кафки и коэффициент репликации для вашего кода.
Для меня я держу 3 как реплики и 3 как фактор репликации.
1
Amit Ahuja
5 Мар 2019 в 09:39
Решение для меня заключалось в том, что я должен был убедиться, что KAFKA_ADVERTISED_HOST_NAME — правильный IP-адрес сервера.
0
Mauvis Ledford
7 Апр 2020 в 17:34
Мы столкнулись с той же проблемой и в производстве. Код работал долго, вдруг мы получили это исключение.
Мы проанализировали, что в коде нет проблем. Поэтому мы попросили группу развертывания перезапустить zookeeper. Перезапуск решил проблему.
-1
Zoe
13 Май 2018 в 12:02
Background: Kafka is used in the recently used project and needs to be integrated with kerberos. Kerberos authentication has been passed, but the data has not been pulled during consumption, and then the shell command is used to consume the data on the server, and the result is the same.
I changed the log level later and found that the following log was printed in a loop:
2020-05-21 10:05:06.651 [ForkJoinPool-1-worker-5] DEBUG org.apache.kafka.clients.Metadata - Updated cluster metadata version 133 to Cluster(id = YFJdX6kITzuEPD3F4zdZEw, nodes = [master:9092 (id: 39 rack: null)], partitions = [Partition(topic = fuxi_test, partition = 0, leader = 39, replicas = [39], isr = [39], offlineReplicas = [])], controller = master:9092 (id: 39 rack: null))
2020-05-21 10:05:06.651 [ForkJoinPool-1-worker-5] DEBUG o.a.k.c.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=test_consumer] Sending FindCoordinator request to broker master:9092 (id: 39 rack: null)
2020-05-21 10:05:06.655 [ForkJoinPool-1-worker-5] DEBUG o.a.k.c.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=test_consumer] Received FindCoordinator response ClientResponse(receivedTimeMs=1590026706655, latencyMs=4, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=2, clientId=consumer-1, correlationId=265), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1 rack: null)))
2020-05-21 10:05:06.655 [ForkJoinPool-1-worker-5] DEBUG o.a.k.c.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=test_consumer] Group coordinator lookup failed: The coordinator is not available.
2020-05-21 10:05:06.655 [ForkJoinPool-1-worker-5] DEBUG o.a.k.c.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-1, groupId=test_consumer] Coordinator discovery failed, refreshing metadata
Finally, I checked online information and found that a child’s shoe said:
__consumer_offsets Does this topic exist
It turns out that the topic really doesn’t exist. Create it manually
kafka-topics --zookeeper master:2181 --partitions 1 --replication-factor 1 --create --topic __consumer_offsets
Restart kafka ok
I have the following structure:
zookeeper: 3.4.12
kafka: kafka_2.11-1.1.0
server1: zookeeper + kafka
server2: zookeeper + kafka
server3: zookeeper + kafka
Created topic with replication factor 3 and partitions 3 by kafka-topics shell script.
./kafka-topics.sh --create --zookeeper localhost:2181 --topic test-flow --partitions 3 --replication-factor 3
And use group localConsumers. it works fine when leader is ok.
./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-flow
Topic:test-flow PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test-flow Partition: 0 Leader: 3 Replicas: 3,2,1 Isr: 3,2,1
Topic: test-flow Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,3,2
Topic: test-flow Partition: 2 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3
Consumers’ log
Received FindCoordinator response ClientResponse(receivedTimeMs=1529508772673, latencyMs=217, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1, clientId=consumer-1, correlationId=0), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=NONE, node=myserver3:9092 (id: 3 rack: null)))
But if leader is down — I get the error in consumer (systemctl stop kafka):
Node 3 is unavailable. ok
./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-flow
Topic:test-flow PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test-flow Partition: 0 Leader: 2 Replicas: 3,2,1 Isr: 2,1
Topic: test-flow Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,2
Topic: test-flow Partition: 2 Leader: 2 Replicas: 2,1,3 Isr: 2,1
Consumers’ log
Received FindCoordinator response
ClientResponse(receivedTimeMs=1529507314193, latencyMs=36,
disconnected=false,
requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1,
clientId=consumer-1, correlationId=149),
responseBody=FindCoordinatorResponse(throttleTimeMs=0,
errorMessage='null', error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1
rack: null)))
- Group coordinator lookup failed: The coordinator is not available.
- Coordinator discovery failed, refreshing metadata
Consumer unable to connect until leader is down or reconnect with another consumer group.
Can’t understand why it happens?
Consumer should be rebalanced to another broker, but it doesn’t.