我试着用卡夫卡。 所有配置都正确完成,但当我试图从控制台产生消息时,我一直得到以下错误
WARN Error while fetching metadata with correlation id 39 :
{4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
Kafka版本:2.11-0.9.0.0
我试着用卡夫卡。 所有配置都正确完成,但当我试图从控制台产生消息时,我一直得到以下错误
WARN Error while fetching metadata with correlation id 39 :
{4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
Kafka版本:2.11-0.9.0.0
当前回答
如果你得到这样重复的错误消息:
Error while fetching metadata with correlation id 3991 : {your.topic=LEADER_NOT_AVAILABLE}
Or
Discovered group coordinator 172.25.1.2:9092 (id: 2147483645 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:677)
(Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:509)
Group coordinator 172.25.1.2:9092 (id: 2147483645 rack: null) is unavailable or invalid, will attempt rediscovery (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:729)
Discovered group coordinator 172.25.40.219:9092 (id: 2147483645 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:677)
Group coordinator 172.25.1.2:9092 (id: 2147483645 rack: null) is unavailable or invalid, will attempt rediscovery (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:729)
然后,你需要像这样在kafka server.properties中配置监听器:
listeners=PLAINTEXT://your.server.ip:9092
这是在Kafka 2.5.0和confluent 5.4.1平台上尝试的解决方案。
其他回答
这个警告(在0.10.2.1中)的另一种可能是,你试图对一个刚刚创建的主题进行投票,而这个主题分区的领导者还不可用,你正在进行领导者选举。
在主题创建和轮询之间等待一秒钟是一种变通方法。
因为我想让我的kafka代理连接到远程生产者和消费者,所以我不想做广告。要注释掉的侦听器。在我的情况下,(在kubernetes上运行kafka),我发现我的kafka pod没有分配任何集群IP。通过从服务中删除行clusterIP: None。呵呵,kubernetes给kafka pod分配了一个内部ip。这解决了我的LEADER_NOT_AVAILABLE问题,也解决了kafka生产者/消费者的远程连接问题。
如果你在本地机器上运行kafka,尝试更新$KAFKA_DIR/config/server。以下行包含的属性: //localhost:9092然后重启kafka。
对于任何试图在kubernetes上运行kafka并遇到这个错误的人来说,这是最终为我解决的问题:
你必须:
将主机名添加到pod规范中,这样kafka就可以找到自己。
or
如果使用hostPort,则需要hostNetwork: true和dnsPolicy: ClusterFirstWithHostNet
这样做的原因是因为Kafka需要与自身对话,它决定使用“广告”侦听器/主机名来找到自己,而不是使用localhost。 即使您有一个将所广告的主机名指向pod的Service,它在pod中也不可见。我真的不知道为什么会这样,但至少有一个变通办法。
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
name: zookeeper-cluster1
namespace: default
labels:
app: zookeeper-cluster1
spec:
replicas: 1
selector:
matchLabels:
app: zookeeper-cluster1
template:
metadata:
labels:
name: zookeeper-cluster1
app: zookeeper-cluster1
spec:
hostname: zookeeper-cluster1
containers:
- name: zookeeper-cluster1
image: wurstmeister/zookeeper:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 2181
- containerPort: 2888
- containerPort: 3888
---
apiVersion: v1
kind: Service
metadata:
name: zookeeper-cluster1
namespace: default
labels:
app: zookeeper-cluster1
spec:
type: NodePort
selector:
app: zookeeper-cluster1
ports:
- name: zookeeper-cluster1
protocol: TCP
port: 2181
targetPort: 2181
- name: zookeeper-follower-cluster1
protocol: TCP
port: 2888
targetPort: 2888
- name: zookeeper-leader-cluster1
protocol: TCP
port: 3888
targetPort: 3888
---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
name: kafka-cluster
namespace: default
labels:
app: kafka-cluster
spec:
replicas: 1
selector:
matchLabels:
app: kafka-cluster
template:
metadata:
labels:
name: kafka-cluster
app: kafka-cluster
spec:
hostname: kafka-cluster
containers:
- name: kafka-cluster
image: wurstmeister/kafka:latest
imagePullPolicy: IfNotPresent
env:
- name: KAFKA_ADVERTISED_LISTENERS
value: PLAINTEXT://kafka-cluster:9092
- name: KAFKA_ZOOKEEPER_CONNECT
value: zookeeper-cluster1:2181
ports:
- containerPort: 9092
---
apiVersion: v1
kind: Service
metadata:
name: kafka-cluster
namespace: default
labels:
app: kafka-cluster
spec:
type: NodePort
selector:
app: kafka-cluster
ports:
- name: kafka-cluster
protocol: TCP
port: 9092
targetPort: 9092
我尝试了这里列出的所有建议。对我有用的是去服务器。属性和添加:
port = 9092
advertised.host.name = localhost
将监听器和advertised_listeners注释掉。