Я настроил своего продюсера 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), чтобы разрешить удаленный доступ ... или я делаю что-то не так? Пожалуйста, помогите.