如果你得到这样重复的错误消息:

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注释掉。