Leader niedostępny Kafka w Console Producer

Próbuję użyć Kafki.
Wszystkie konfiguracje są wykonane poprawnie, ale kiedy próbuję wygenerować komunikat z konsoli, pojawia się następujący błąd

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

Wersja Kafki: 2.11-0.9.0.0

Może to być związane z advertised.host.nameustawieniem w Twoim server.properties.

Co może się zdarzyć, że producent próbuje dowiedzieć się, kto jest liderem na danej partycji, dane swoje advertised.host.namei advertised.portpróbuje się połączyć. Jeśli te ustawienia nie są poprawnie skonfigurowane, może pomyśleć, że lider jest niedostępny.

Wypróbowałem wszystkie wymienione tutaj zalecenia. Udało mi się przejść do server.propertiesi dodać:

port = 9092
advertised.host.name = localhost 

Zostaw listenersi advertised_listenersskomentuj.

Rozwiązaniem dla mnie jest takie ustawienie słuchaczy:

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

To sprawia, że ​​broker KAFKA nasłuchuje wszystkich interfejsów.

Miałem kafkę działającą jako kontener Dockera i podobne wiadomości były zalewane do dziennika.
I KAFKA_ADVERTISED_HOST_NAMEzostał ustawiony na „kafka”.

W moim przypadku przyczyną błędu był brak /etc/hostsrekordu „kafka” w samym kontenerze „kafka”.
Na przykład uruchomienie ping kafkaw kontenerze „kafka” nie powiedzie sięping: bad address 'kafka'

W przypadku Dockera problem ten został rozwiązany poprzez określenie hostnamekontenera.

Możliwości osiągnięcia tego celu:

• docker run --hostname ...
• docker run -it --add-host ...
• hostname w docker-compose
• hostname w definicji zadania AWS EC2

Używam kafka_2.12-0.10.2.1:

vi config/server.properties

dodaj poniższy wiersz:

listeners=PLAINTEXT://localhost:9092

• Nie ma potrzeby zmieniania anonsed.listeners, ponieważ pobiera on wartość z właściwości listener std.

Nazwa hosta i port, które broker będzie reklamował producentom i konsumentom. Jeśli nie jest ustawiony,

• używa wartości dla „detektorów”, jeśli jest skonfigurowana

W przeciwnym razie użyje wartości zwróconej z java.net.InetAddress.getCanonicalHostName().

zatrzymaj brokera Kafka:

bin/kafka-server-stop.sh

zrestartuj brokera:

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

a teraz nie powinieneś widzieć żadnych problemów.

Byłem świadkiem tego samego problemu w ciągu ostatnich 2 tygodni podczas pracy z Kafką i od tamtej pory czytam ten post Stackoverflow.

Po 2 tygodniach analizy wywnioskowałem, że w moim przypadku dzieje się tak, gdy próbuję napisać wiadomość na temat, który nie istnieje .

Rezultat w moim przypadku jest taki, że Kafka odsyła komunikat o błędzie, ale jednocześnie tworzy temat, który wcześniej nie istniał. Więc jeśli po tym zdarzeniu spróbuję ponownie wygenerować jakąkolwiek wiadomość do tego tematu, błąd nie będzie się już pojawiał jako temat, który został utworzony.

UWAGA: Możliwe, że moja konkretna instalacja platformy Kafka została skonfigurowana tak, aby automatycznie tworzyła temat, gdy ten sam nie istnieje; To powinno wyjaśniać, dlaczego w moim przypadku widzę problem tylko raz dla każdego tematu po zresetowaniu tematów: Twoja konfiguracja może być inna iw takim przypadku ciągle otrzymywałbyś ten sam błąd.

Pozdrowienia,

Luca Tampellini

Zwykle otrzymujemy tę wiadomość, gdy próbujemy zasubskrybować temat, który nie został jeszcze utworzony. Generalnie polegamy na tematach, które mają być tworzone a priori w naszych wdrożonych środowiskach, ale mamy testy komponentów, które działają na zadokeryzowanej instancji kafka, która za każdym razem uruchamia się czysto.

W takim przypadku używamy AdminUtils w naszej konfiguracji testowej, aby sprawdzić, czy temat istnieje, i utworzyć go, jeśli nie. Zobacz inne przepełnienie stosu, aby dowiedzieć się więcej o konfigurowaniu AdminUtils.

Inną możliwością dla tego ostrzeżenia (w 0.10.2.1) jest to, że próbujesz przeprowadzić ankietę na temat, który właśnie został utworzony, a lider tego podziału tematycznego nie jest jeszcze dostępny, jesteś w środku wyborów przywódczych.

Aby obejść ten problem, należy poczekać sekundę między utworzeniem tematu a odpytywaniem.

Dla każdego, kto próbował uruchomić kafkę na kubernetes i napotkał ten błąd, oto co ostatecznie rozwiązało to dla mnie:

Musisz albo:

1. Dodaj hostnamedo specyfikacji kapsuły, w ten sposób kafka może się znaleźć.

lub

2. Jeśli używasz hostPort, potrzebujesz hostNetwork: trueidnsPolicy: ClusterFirstWithHostNet

Powodem tego jest to, że Kafka musi rozmawiać ze sobą i decyduje się użyć „reklamowanego” odbiornika / nazwy hosta, aby się znaleźć, zamiast używać hosta lokalnego. Nawet jeśli masz usługę, która wskazuje reklamowaną nazwę hosta na kapsułę, nie jest ona widoczna z jej wnętrza. Naprawdę nie wiem, dlaczego tak jest, ale przynajmniej istnieje obejście tego problemu.

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

Dodając to, ponieważ może to pomóc innym. Częstym problemem może być błędna konfiguracja advertised.host.name. Z Dockerem przy użyciu docker-compose ustawienie nazwy usługi wewnątrz KAFKA_ADVERTISED_HOST_NAMEnie zadziała, chyba że ustawisz również nazwę hosta. docker-compose.ymlprzykład:

  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

Powyższe bez hostname: kafkamoże powodować problemy LEADER_NOT_AVAILABLEpodczas próby połączenia. Można znaleźć przykład roboczej docker-composekonfiguracji tutaj

W moim przypadku działało dobrze w domu, ale zawodziło w biurze, w momencie, gdy łączę się z siecią biurową.

Zmodyfikowano więc config / server.properties listeners = PLAINTEXT: //: 9092 to listeners = PLAINTEXT: // localhost: 9092

W moim przypadku zrozumiałem, opisując grupę konsumentów

Jeśli używasz kafki na komputerze lokalnym, spróbuj zaktualizować $ KAFKA_DIR / config / server.properties za pomocą poniższej linii:, listeners=PLAINTEXT://localhost:9092a następnie zrestartuj kafkę.

Używam docker-compose do zbudowania kontenera Kafka przy użyciu wurstmeister/kafkaobrazu. Dodanie KAFKA_ADVERTISED_PORT: 9092właściwości do mojego docker-composepliku rozwiązało ten błąd.

Ponieważ chciałem, aby mój broker kafka łączył się ze zdalnymi producentami i konsumentami, więc nie chcę advertised.listenerbyć komentowany. W moim przypadku (uruchamiając kafkę na kubernetes), dowiedziałem się, że moja pod kafka nie ma przypisanego adresu IP klastra. Usuwając linię clusterIP: Nonez services.yml, kubernetes przypisuje wewnętrzny adres IP do kafki pod. To rozwiązało mój problem z LEADER_NOT_AVAILABLE, a także zdalnym połączeniem producentów / konsumentów kafka.

Gdy pojawi się błąd LEADER_NOT_AVAILABLE, po prostu zrestartuj brokera kafka:

/bin/kafka-server-stop.sh

śledzony przez

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

(Uwaga: Zookeeper musi działać do tego czasu, jeśli zrobisz to w inny sposób, to nie zadziała)

Poniższa linia, którą dodałem config/server.properties, rozwiązała mój problem, podobnie jak powyższy problem. Mam nadzieję, że to pomoże, jest dobrze udokumentowane w pliku server.properties, spróbuj przeczytać i zrozumieć, zanim to zmodyfikujesz. advertised.listeners=PLAINTEXT://<your_kafka_server_ip>:9092

Dla wszystkich, którzy zmagają się z konfiguracją SSL Kafka i widzą ten błąd LEADER_NOT_AVAILABLE. Jednym z powodów, dla których może się to zepsuć, jest magazyn kluczy i magazyn zaufanych certyfikatów. W magazynie kluczy musisz mieć klucz prywatny serwera + podpisany certyfikat serwera. W magazynie zaufanych certyfikatów klienta musisz mieć pośredni certyfikat CA, aby klient mógł uwierzytelniać serwer kafka. Jeśli do komunikacji między brokerem będzie używany protokół SSL, ten magazyn zaufanych certyfikatów musi być również ustawiony w pliku server.properties brokerów, aby mogli się wzajemnie uwierzytelniać.

Ten ostatni element, którego omyłkowo przegapiłem, spowodował wiele bolesnych godzin, aby dowiedzieć się, co może oznaczać ten błąd LEADER_NOT_AVAILABLE. Mam nadzieję, że to może komuś pomóc.

Problem został rozwiązany po dodaniu ustawienia nasłuchiwania w pliku server.properties znajdującym się w katalogu config. Listeners = PLAINTEXT: // localhost (lub twój serwer): 9092 Zrestartuj kafkę po tej zmianie. Używana wersja 2.11

U mnie było to spowodowane błędną konfiguracją
portu Dockera (9093)
Port poleceń Kafka "bin / kafka-console-producer.sh --broker-list localhost: 9092 --topic TopicName"
Sprawdziłem konfigurację, aby pasowała do portu i teraz wszystko jest w porządku

Dla mnie przyczyną było użycie konkretnego Zookeepera, który nie był częścią pakietu Kafka. Ten Zookeeper został już zainstalowany na maszynie do innych celów. Najwyraźniej Kafka nie współpracuje z żadnym Zookeeperem. Przejście na Zookeeper, który był dostarczony z Kafką, rozwiązało to za mnie. Aby nie kolidować z istniejącym Zookeeperem, musiałem zmodyfikować moją konfigurację, aby Zookeeper nasłuchiwał na innym porcie:

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

Reklamowani słuchacze, jak wspomniano w powyższych odpowiedziach, mogą być jednym z powodów. Inne możliwe przyczyny to:

1. Temat mógł nie zostać utworzony. Możesz to sprawdzić za pomocąbin/kafka-topics --list --zookeeper <zookeeper_ip>:<zookeeper_port>
2. Sprawdź serwery bootstrap, które przekazałeś producentowi, aby pobrać metadane. Jeśli serwer startowy nie zawiera najnowszych metadanych dotyczących tematu (na przykład, gdy utracił roszczenie zookeeper). Musisz dodać więcej niż jeden serwer startowy.

Upewnij się również, że reklamowany odbiornik jest ustawiony na IP:9092zamiastlocalhost:9092 . To ostatnie oznacza, że ​​broker jest dostępny tylko za pośrednictwem hosta lokalnego.

Kiedy napotkałem błąd, pamiętam, że użyłem PLAINTEXT://<ip>:<PORT>listy serwerów bootstrap (lub listy brokerów) i dziwnie zadziałało.

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

Dla mnie nie podałem identyfikatora brokera dla instancji Kafka. Czasami otrzyma nowy identyfikator od zookeepera po ponownym uruchomieniu w środowisku Docker. Jeśli twój identyfikator brokera jest większy niż 1000, po prostu określ zmienną środowiskową KAFKA_BROKER_ID.

Użyj tego, aby zobaczyć brokerów, tematy i partycje.

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

Wiem, że to zostało opublikowane dawno temu, chciałbym podzielić się tym, jak go rozwiązałem.
ponieważ mam laptopa w biurze ( skonfigurowano VPN i proxy ).
sprawdziłem zmienną środowiskową NO_PROXY

> echo %NO_PROXY%

zwrócił z pustymi wartościami,
teraz ustawiłem NO_PROXY z localhost i 127.0.0.1

> set NO_PROXY=127.0.0.1,localhost  

jeśli chcesz dołączyć do istniejących wartości, to

> set NO_PROXY=%NO_PROXY%,127.0.0.1,localhost  

po tym zrestartowałem zookeepera i kafka
działała jak urok