apache kafka - potenciales - Los metadatos de captación de clientes de Kafka para temas fracasaron

estrategias para captar clientes potenciales pdf

Question

Estoy intentando escribir un cliente Java para los servidores Kafka y ZooKeeper de un tercero. Puedo enumerar y describir temas, pero cuando intento leer alguno, se ClosedChannelException una ClosedChannelException . Los reproduzco aquí con el cliente de línea de comando.

$ bin/kafka-console-consumer.sh --zookeeper 255.255.255.255:2181 --topic eventbustopic
[2015-06-02 16:23:04,375] WARN Fetching topic metadata with correlation id 0 for topics [Set(eventbustopic)] from broker [id:1,host:SOME_HOST,port:9092] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException                                       
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)           
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)        
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:113)                
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)        
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93)        
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)         
[2015-06-02 16:23:04,515] WARN Fetching topic metadata with correlation id 0 for topics [Set(eventbustopic)] from broker [id:0,host:SOME_HOST,port:9092] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException                                       
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)           
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)        
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:113)                
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)        
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93)        
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)         

Los comandos alternativos tienen éxito:

$ bin/kafka-topics.sh --describe --zookeeper 255.255.255.255:2181 --topic eventbustopic
Topic:eventbustopic   PartitionCount:2        ReplicationFactor:1     Configs:
    Topic: eventbustopic  Partition: 0    Leader: 1       Replicas: 1     Isr: 1
    Topic: eventbustopic  Partition: 1    Leader: 0       Replicas: 0     Isr: 0

$ bin/kafka-topics.sh --list --zookeeper 255.255.255.255:2181 --topic eventbustopic
eventbustopic

(Los ips fueron redactados y reemplazados por 255.255.255.255)

Cuando busco esta excepción, veo problemas en el lado del productor, de hecho, la fuente de ClientUtils.fetchTopicMetadata sugiere que esto es utilizado principalmente por los productores.

Una preocupación que tengo es que esto podría ser un producto del diseño de la red: los paquetes son destrozados por Haproxy y enviados a través de una VPN.

¿Qué está trabajando exactamente aquí?

Answer 1

El intermediario le dice al cliente qué nombre de host debe usarse para producir / consumir mensajes. Por defecto, Kafka usa el nombre de host del sistema en el que se ejecuta. Si este nombre de host no se puede resolver por el lado del cliente, se obtiene esta excepción.

Puede intentar configurar advertised.host.name en la configuración de Kafka en un nombre de host / dirección que los clientes deberían usar.

Answer 2

Esta es mi manera de resolver este problema:

1. ejecute bin/kafka-server-stop.sh para detener la ejecución del servidor kafka.
2. modifique el archivo de propiedades config/server.properties agregando una línea: listeners=PLAINTEXT://{ip.of.your.kafka.server}:9092
3. reinicie el servidor kafka.

Como sin la configuración de lisener, kafka usará java.net.InetAddress.getCanonicalHostName() para obtener la dirección que escucha el servidor de socket.

Answer 3

Se encontró con este error en AWS. El problema fue que era demasiado restrictivo con el grupo de seguridad y establecí los puertos 2181 y 9092 en "mi IP". Esto significaba que la instancia kafka no podía encontrar el ZK ejecutándose en la misma caja.

Solución - ábrela - un poco.

Answer 4

Usted tiene un problema con Zookeeper. 255.255.255.255:2181 no es una dirección válida de Zookeeper; esta es una dirección de difusión en su red o una máscara de subred. Para que las cosas funcionen, busque la dirección IP o el nombre de host de la máquina que ejecuta Zookeeper.