diff --git a/src/common/tools/kafka/Variables.py b/src/common/tools/kafka/Variables.py index 89ac42f90562c1ff66525b6fd45bf4463a8dc68d..9abc32b3e10a71c6191f987c26459855381fe6b7 100644 --- a/src/common/tools/kafka/Variables.py +++ b/src/common/tools/kafka/Variables.py @@ -20,14 +20,14 @@ from common.Settings import get_setting LOGGER = logging.getLogger(__name__) -KFK_SERVER_ADDRESS_TEMPLATE = 'kafka-service.{:s}.svc.cluster.local:{:s}' class KafkaConfig(Enum): - KFK_NAMESPACE = get_setting('KFK_NAMESPACE') - KFK_PORT = get_setting('KFK_SERVER_PORT') - # SERVER_IP = "127.0.0.1:9092" - server_address = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT) - ADMIN_CLIENT = AdminClient({'bootstrap.servers': server_address}) + KFK_SERVER_ADDRESS_TEMPLATE = 'kafka-service.{:s}.svc.cluster.local:{:s}' + KFK_NAMESPACE = get_setting('KFK_NAMESPACE') + KFK_PORT = get_setting('KFK_SERVER_PORT') + # SERVER_ADDRESS = "127.0.0.1:9092" + SERVER_ADDRESS = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT) + ADMIN_CLIENT = AdminClient({'bootstrap.servers': SERVER_ADDRESS }) class KafkaTopic(Enum): REQUEST = 'topic_request' @@ -42,6 +42,7 @@ class KafkaTopic(Enum): Method to create Kafka topics defined as class members """ all_topics = [member.value for member in KafkaTopic] + LOGGER.debug("Kafka server address is: {:} ".format(KafkaConfig.SERVER_ADDRESS.value)) if( KafkaTopic.create_new_topic_if_not_exists( all_topics )): LOGGER.debug("All topics are created sucsessfully") return True diff --git a/src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py b/src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py index d27de54f3cddfd0d70d656a89c45adc50e518289..1559457d7fc80c3e9718c182f736b3f44abdeaea 100644 --- a/src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py +++ b/src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py @@ -38,7 +38,7 @@ class KpiValueApiServiceServicerImpl(KpiValueAPIServiceServicer): ) -> Empty: LOGGER.debug('StoreKpiValues: Received gRPC message object: {:}'.format(request)) producer_obj = KafkaProducer({ - 'bootstrap.servers' : KafkaConfig.SERVER_IP.value + 'bootstrap.servers' : KafkaConfig.SERVER_ADDRESS.value }) for kpi_value in request.kpi_value_list: kpi_value_to_produce : Tuple [str, Any, Any] = ( diff --git a/src/kpi_value_writer/service/KpiValueWriter.py b/src/kpi_value_writer/service/KpiValueWriter.py index 022126fd0f732d6e7b52671a707385c30b9f052c..5e2b6babe210e4fb71eb8617432a6dfd5d164407 100644 --- a/src/kpi_value_writer/service/KpiValueWriter.py +++ b/src/kpi_value_writer/service/KpiValueWriter.py @@ -51,10 +51,10 @@ class KpiValueWriter(GenericGrpcService): metric_writer = MetricWriterToPrometheus() kafka_consumer = KafkaConsumer( - { 'bootstrap.servers' : KafkaConfig.SERVER_IP.value, + { 'bootstrap.servers' : KafkaConfig.SERVER_ADDRESS.value, 'group.id' : __class__, 'auto.offset.reset' : 'latest'} - ) + ) kafka_consumer.subscribe([KafkaTopic.VALUE.value]) LOGGER.debug("Kafka Consumer start listenng on topic: {:}".format(KafkaTopic.VALUE.value)) print("Kafka Consumer start listenng on topic: {:}".format(KafkaTopic.VALUE.value))