Не удается использовать события kafka удаленно. Ошибка: connect ECONNREFUSED 5.6.7.8:9092 - PullRequest
Новая
       11

Не удается использовать события kafka удаленно. Ошибка: connect ECONNREFUSED 5.6.7.8:9092

0 голосов
/ 04 мая 2018

Я настроил своего продюсера kafka заново, следуя инструкции, упомянутой здесь: https://www.digitalocean.com/community/tutorials/how-to-install-apache-kafka-on-ubuntu-14-04

Я передаю некоторые события производителю, используя cron и следующий скрипт на сервере с IP: 1.2.3.4 

#!/usr/bin/env python
import threading, logging, time
import multiprocessing
import requests
import datetime
import json
from kafka import KafkaProducer

class CheckApis():
    apisList = {"a": "https://test.eng.com/"}
    kafkaProducer = "1.2.3.4:9092"
    kafkaTopic = "sometopic"
    producer = KafkaProducer(bootstrap_servers=kafkaProducer)
    for key, value in apisList.items():
        headers = {};
        response = requests.request("GET", value, headers=headers)
        message = {"app_name": key, "status": response.status_code, "message": "none", "timestamp": str(datetime.datetime.utcnow())}
        producer.send(kafkaTopic, json.dumps(message).encode('utf-8'));
        print (response.text)
        print (response.status_code)
    producer.close()

Это хорошо работает, и я могу увидеть события, отправленные с помощью этой команды: 

~/kafka/bin/kafka-console-consumer.sh --zookeeper 1.2.3.4:2181 --topic sometopic --from-beginning

Но когда я пытаюсь использовать эти события удаленно с другого сервера (моего ноутбука), происходит сбой с ошибкой:

ошибка: произошла ошибка при прослушивании событий kafka Ошибка: connect ECONNREFUSED 5.6.7.8:9092 (некоторые другие IP здесь не 1.2.3.4)

Вот мой потребительский код (в узле js, использующем kafka-node): 

var ConsumerGroup = require('kafka-node').ConsumerGroup;
    var healthConsumerOption = {
        host: '1.2.3.4:2181',
        autoCommit: true,
        groupId: os.hostname(),
        sessionTimeout: 15000,
        protocol: ['roundrobin'],
        fromOffset: 'latest'
    };
    var healthConsumerGroup = new ConsumerGroup(healthConsumerOption, healthTopics);
        listenHealthEventsKafka: function(connections){
            try{
                healthConsumerGroup.on('error', onError);
                healthConsumerGroup.on('message', onMessage);
                healthConsumerGroup.on('connect', function(){
                    logger.info("Health consumer group is ready. ")
                });

                function onMessage(message){
                    var jsonData = JSON.parse(message.value);
                    console.log(message);
                };

                function onError(error){
                    logger.error("Some error occured while listening to kafka events " +error);
                }

                process.once('SIGNINT', function(){
                    async.each([healthConsumerGroup], function(consumer, callback){
                        logger.info("Closing the kafka health consumer process ");
                        consumer.close(true, callback);
                    });
                })
            }catch(error){
                logger.error("Could not connect to kafka events for build " +error);
            }
        }

Нужно ли выполнять дополнительную настройку на сервере Kafka (server.properties), чтобы разрешить удаленный доступ ... или я делаю что-то не так? Пожалуйста, помогите.

1 Ответ

0 голосов
/ 04 мая 2018

На вашем брокере Kafka server.properties необходимо установить advertised.listeners на внешний IP , чтобы клиенты могли правильно подключиться к нему. В противном случае они попытаются подключиться к внутреннему IP-адресу (поскольку advertised.listeners по умолчанию будет listeners, если явно не установлено)

Ссылка: https://kafka.apache.org/documentation/#brokerconfigs

...