Лидер не доступен Кафка в консоли производителя

Я пытаюсь использовать Кафку.
Все настройки выполнены правильно, но когда я пытаюсь создать сообщение с консоли, я получаю следующую ошибку

WARN Error while fetching metadata with correlation id 39 : 
     {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

Кафка версия: 2.11-0.9.0.0

Это может быть связано с advertised.host.nameнастройкой в ​​вашем server.properties.

Что может случиться, что ваш продюсер пытается выяснить , кто является лидером данного раздела, выясняет его advertised.host.nameи advertised.portи пытается подключиться. Если эти настройки не настроены правильно, он может подумать, что лидер недоступен.

Я перепробовал все рекомендации, перечисленные здесь. То, что работало для меня, было пойти server.propertiesи добавить:

port = 9092
advertised.host.name = localhost 

Оставьте listenersи advertised_listenersзакомментируйте.

Что для меня решило, так это настроить слушателей так:

advertised.listeners = PLAINTEXT://my.public.ip:9092
listeners = PLAINTEXT://0.0.0.0:9092

Это заставляет брокера KAFKA слушать все интерфейсы.

У меня была Кафка, работающая как контейнер Docker, и подобные сообщения попадали в журнал.
И KAFKA_ADVERTISED_HOST_NAMEбыл установлен на «Кафка».

В моем случае причиной ошибки была отсутствующая /etc/hostsзапись для «kafka» в самом контейнере «kafka».
Так, например, запуск ping kafkaвнутри контейнера 'kafka' завершится неудачно сping: bad address 'kafka'

В терминах Docker эта проблема решается путем указания hostnameконтейнера.

Варианты достижения этого:

• docker run --hostname ...
• docker run -it --add-host ...
• hostname в докер-сочинении
• hostname в определении задачи AWS EC2

Я использую kafka_2.12-0.10.2.1:

vi config/server.properties

добавить строку ниже:

listeners=PLAINTEXT://localhost:9092

• Нет необходимости изменять Advanised.listeners, поскольку он выбирает значение из свойства слушателя std.

Имя хоста и порт брокер будет рекламировать производителям и потребителям. Если не установлено,

• он использует значение для «слушателей», если настроено

В противном случае он будет использовать значение, возвращенное из java.net.InetAddress.getCanonicalHostName().

остановить брокера Kafka:

bin/kafka-server-stop.sh

перезапустить брокера:

bin/kafka-server-start.sh -daemon config/server.properties

и теперь вы не должны видеть никаких проблем.

Я был свидетелем этой же проблемы в последние 2 недели, работая с Кафкой, и с тех пор читаю этот пост Stackoverflow.

После 2 недель анализа я пришел к выводу, что в моем случае это происходит при попытке создать сообщения на тему, которая не существует .

Результатом в моем случае является то, что Кафка отправляет сообщение об ошибке обратно, но в то же время создает тему, которой раньше не было. Поэтому, если я попытаюсь снова создать какое-либо сообщение для этой темы после этого события, ошибка больше не будет отображаться как созданная тема.

ПОЖАЛУЙСТА, ОБРАТИТЕ ВНИМАНИЕ: Возможно, моя конкретная установка Kafka была настроена на автоматическое создание темы, когда ее нет; это должно объяснить, почему в моем случае я могу видеть проблему только один раз для каждой темы после сброса тем: ваша конфигурация может отличаться, и в этом случае вы будете снова и снова получать одну и ту же ошибку.

С Уважением,

Лука Тампеллини

Мы склонны получать это сообщение, когда пытаемся подписаться на тему, которая еще не была создана. Как правило, мы полагаемся на темы, которые создаются априори в наших развернутых средах, но у нас есть тесты компонентов, которые запускаются на докеризованном экземпляре kafka, который каждый раз запускается чистым.

В этом случае мы используем AdminUtils в нашей тестовой настройке, чтобы проверить, существует ли тема, и создать ее, если нет. Посмотрите это другое переполнение стека для получения дополнительной информации о настройке AdminUtils.

Другая возможность для этого предупреждения (в 0.10.2.1) состоит в том, что вы пытаетесь опросить тему, которая была только что создана, и лидер этого тематического раздела еще не доступен, вы находитесь в середине выборов лидера.

Ожидание секунды между созданием темы и опросом - это обходной путь.

Любой, кто пытается запустить kafka на kubernetes и сталкивается с этой ошибкой, вот что окончательно решило ее для меня:

Вы должны либо:

1. Добавьте hostnameк спецификации стручка, чтобы кафка могла найти себя.

или

2. Если используете hostPort, то вам нужно hostNetwork: trueиdnsPolicy: ClusterFirstWithHostNet

Причина этого в том, что Кафке нужно поговорить с самим собой, и он решает использовать «объявленное» имя слушателя / хоста, чтобы найти себя, а не использовать localhost. Даже если у вас есть Служба, которая указывает рекламируемое имя хоста на модуль, оно не отображается внутри модуля. Я действительно не знаю, почему это так, но по крайней мере есть обходной путь.

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

Добавление этого, так как это может помочь другим. Распространенной проблемой может быть неправильная настройка advertised.host.name. С помощью Docker, использующего docker-compose, установка имени службы внутри не KAFKA_ADVERTISED_HOST_NAMEбудет работать, если вы не зададите и имя хоста. docker-compose.ymlпример:

  kafka:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    hostname: kafka
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

Вышеуказанное без hostname: kafkaможет выдать LEADER_NOT_AVAILABLEпри попытке подключения. Вы можете найти пример рабочей docker-composeконфигурации здесь

В моем случае все работало нормально дома, но в офисе не получалось, когда я подключался к офисной сети.

Поэтому изменил config / server.properties listeners = PLAINTEXT: //: 9092 to listeners = PLAINTEXT: // localhost: 9092

В моем случае я получал, когда описывал Consumer Group

Если вы запускаете kafka на локальном компьютере, попробуйте обновить $ KAFKA_DIR / config / server.properties с помощью строки ниже: listeners=PLAINTEXT://localhost:9092и затем перезапустите kafka.

Я использую docker-compose для сборки контейнера Kafka с использованием wurstmeister/kafkaimage. Добавление KAFKA_ADVERTISED_PORT: 9092свойства в мой docker-composeфайл решило эту ошибку для меня.

Поскольку я хотел, чтобы мой брокер kafka связывался с удаленными производителями и потребителями, я не хочу, advertised.listenerчтобы его комментировали. В моем случае (при запуске kafka на kubernetes) я обнаружил, что моему модулю kafka не был назначен какой-либо IP-адрес кластера. Удаляя строку clusterIP: Noneиз services.yml, kubernetes назначает внутренний ip для модуля kafka. Это решило мою проблему LEADER_NOT_AVAILABLE, а также удаленное подключение производителей / потребителей кафки.

Когда выдается ошибка LEADER_NOT_AVAILABLE, просто перезапустите брокер kafka:

/bin/kafka-server-stop.sh

с последующим

/bin/kafka-server-start.sh config/server.properties

(Примечание: Zookeeper должен быть запущен к этому времени, если вы делаете иначе, он не будет работать)

Эта строка ниже, я добавил config/server.properties, что решил мою проблему, аналогичную вышеупомянутой проблеме. Надеюсь, это поможет, это довольно хорошо задокументировано в файле server.properties, попробуйте прочитать и понять, прежде чем изменять это. advertised.listeners=PLAINTEXT://<your_kafka_server_ip>:9092

Для всех, кто борется с настройкой Kafka ssl и видит эту ошибку LEADER_NOT_AVAILABLE. Одной из причин, которые могут быть повреждены, является хранилище ключей и хранилище доверенных сертификатов. В хранилище ключей необходимо иметь закрытый ключ сервера + подписанный сертификат сервера. В клиентском хранилище доверенных сертификатов необходимо иметь промежуточный сертификат CA, чтобы клиент мог аутентифицировать сервер kafka. Если вы будете использовать ssl для взаимодействия между брокерами, вам необходимо также установить это доверенное хранилище в свойствах server.properties брокеров, чтобы они могли аутентифицировать друг друга.

Этот последний фрагмент, который я по ошибке пропустил, заставил меня много мучительных часов узнать, что может означать эта ошибка LEADER_NOT_AVAILABLE. Надеюсь, это может кому-нибудь помочь.

Проблема устранена после добавления настройки прослушивателя в файл server.properties, расположенный в каталоге config. listeners = PLAINTEXT: // localhost (или ваш сервер): 9092 Перезапустите kafka после этого изменения. Используемая версия 2.11

Для меня это произошло из-за неправильной конфигурации
Docker port (9093)
Командный порт Kafka "bin / kafka-console-producer.sh --broker-list localhost: 9092 --topic TopicName"
Я проверил свою конфигурацию на соответствие порту и Теперь все в порядке

Для меня причиной было использование специального Zookeeper, который не был частью пакета Kafka. Этот Zookeeper уже был установлен на машине для других целей. По всей видимости, Кафка не работает ни с одним Zookeeper. Переход на Zookeeper, поставляемый с Кафкой, решил это для меня. Чтобы не конфликтовать с существующим Zookeeper, мне пришлось изменить конфигурацию, чтобы Zookeeper прослушивал другой порт:

[root@host /opt/kafka/config]# grep 2182 *
server.properties:zookeeper.connect=localhost:2182
zookeeper.properties:clientPort=2182

Рекламируемые слушатели, упомянутые в ответах выше, могут быть одной из причин. Другие возможные причины:

1. Возможно, тема не была создана. Вы можете проверить это используяbin/kafka-topics --list --zookeeper <zookeeper_ip>:<zookeeper_port>
2. Проверьте серверы начальной загрузки, которые вы дали производителю для получения метаданных. Если сервер начальной загрузки не содержит последние метаданные о теме (например, когда он потерял свое требование zookeeper). Вы должны добавить более одного сервера начальной загрузки.

Кроме того, убедитесь, что рекламируемый слушатель установлен IP:9092вместоlocalhost:9092 . Последнее означает, что брокер доступен только через локальный хост.

Когда я столкнулся с ошибкой, я помню, что использовал PLAINTEXT://<ip>:<PORT> в списке серверов начальной загрузки (или список брокеров), и это работало, как ни странно.

bin/kafka-console-producer --topic sample --broker-list PLAINTEXT://<IP>:<PORT>

Для меня я не указал идентификатор брокера для экземпляра Kafka. Иногда он получает новый идентификатор от zookeeper при перезапуске в среде Docker. Если идентификатор вашего брокера больше 1000, просто укажите переменную средыKAFKA_BROKER_ID .

Используйте это, чтобы увидеть брокеров, темы и разделы.

brew install kafkacat
kafkacat -b [kafka_ip]:[kafka_poot] -L

я знаю, что это было опубликовано давным-давно, я хотел бы поделиться, как я решил это.
так как у меня есть мой офисный ноутбук ( VPN и прокси был настроен).
я проверил переменную окружения NO_PROXY

> echo %NO_PROXY%

он вернулся с пустыми значениями
теперь я установил NO_PROXY с localhost и 127.0.0.1

> set NO_PROXY=127.0.0.1,localhost  

если вы хотите добавить к существующим значениям, то

> set NO_PROXY=%NO_PROXY%,127.0.0.1,localhost  

после этого я перезапустил зоопарк и кафка заработала
как шарм