Apache Kafka: Metadaten können nicht aktualisiert werden/java.nio.channels.ClosedChannelException

Question

Ich fange gerade erst an mit Apache Kafka/Zookeeper und habe Probleme beim Einrichten eines Clusters in AWS. Derzeit habe ich drei Server:

Ein laufender Zookeeper und zwei laufende Kafka.

Ich kann die Kafka-Server ohne Problem starten und kann Themen auf beiden erstellen. Allerdings kommt das Problem, wenn ich versuche, einen Produzenten auf einer Maschine und einen Verbraucher auf den anderen Seite zu starten:

auf dem Kafka Produzenten:

kafka-console-producer.sh --broker-list <kafka server 1 aws public dns>:9092,<kafka server 2 aws public dns>:9092 --topic samsa 

auf dem Kafka Verbraucher:

kafka-console-consumer.sh --zookeeper <zookeeper server ip>:2181 --topic samsa 

Ich tippe eine Nachricht auf den Produzenten ("Hallo") und nichts passiert für eine Weile. Dann bekomme ich diese Meldung:

ERROR Error when sending message to topic samsa with key: null, value: 2 bytes 
with error: Failed to update metadata after 60000 ms. 
(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) 

Auf der Verbraucherseite ich diese Nachricht erhalten, die in regelmäßigen Abständen wiederholt:

WARN Fetching topic metadata with correlation id # for topics [Set(samsa)] from broker [BrokerEndPoint(<broker.id>,<producer's advertised.host.name>,9092)] failed (kafka.client.ClientUtils$) 
java.nio.channels.ClosedChannelException 
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110) 
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75) 
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74) 
    at kafka.producer.SyncProducer.send(SyncProducer.scala:119) 
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59) 
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94) 
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66) 
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63) 

Nach einer Weile wird der Hersteller dann beginnt schnell diese Fehlermeldung wirft mit # Erhöhung inkrementell:

WARN Error while fetching metadata with correlation id # : {samsa=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) 

Nicht sicher wohin von hier zu gehen. Lassen Sie mich wissen, wenn weitere Details zu meinen Konfigurationsdateien benötigt werden

Answer 1

Dies war ein Konfigurationsproblem.

Um es laufen mehrere Änderungen an config Dateien zu erhalten passieren musste:

In config/server.properties auf jedem Kafka Server:

• host.name: <Public IP>
• advertised.host.name: <AWS Public DNS Address>

In config/producer.properties auf Jeder Kafka-Server:

• metadata.broker.list: <Producer Server advertised.host.name>:<Producer Server port>,<Consumer Server advertised.host.name>:<Consumer Server port>

in/etc/hosts auf jedem Kafka-Server ändern 127.0.0.1 localhost localhost.localdomain zu:

<Public IP> localhost localhost.localdomain