Commit bdcb5c01 authored by Waleed Akbar's avatar Waleed Akbar
Browse files

Changes to resolve Kafka server error 

- KFK_SERVER_ADDRESS_TEMPLATE now defined inside the class KafkaConfig.
- variable renamed to "SERVER_ADDRESS" from "server_address"
parent 58408ee6

src/common/tools/kafka/Variables.py

+7 −6
Original line number Diff line number Diff line
@@ -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_SERVER_ADDRESS_TEMPLATE = 'kafka-service.{:s}.svc.cluster.local:{:s}'

    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})

    # 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

src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py

+1 −1
Original line number Diff line number Diff line
@@ -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] = (

src/kpi_value_writer/service/KpiValueWriter.py

+2 −2
Original line number Diff line number Diff line
@@ -51,7 +51,7 @@ 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'}

        )