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

Kafka deployment script in gitlab-ci.file (2) 

- Improvements in Kafka.variables files.
- In KpiValueApiServiceImpl corrected the call from "admin_client()" to "kafka_address()"
parent f9e63a88

src/common/tools/kafka/Variables.py

+7 −9
Original line number Diff line number Diff line
@@ -30,14 +30,12 @@ class KafkaConfig(Enum): 
        if kafka_server_address is None:

            KFK_NAMESPACE        = get_setting('KFK_NAMESPACE')

            KFK_PORT             = get_setting('KFK_SERVER_PORT')

            SERVER_ADDRESS = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)

        return SERVER_ADDRESS

            kafka_server_address = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)

        return kafka_server_address

        

    @staticmethod

    def get_admin_client():

        SERVER_ADDRESS = KafkaConfig.get_kafka_address()

        LOGGER.debug("KAFKA_SERVER_ADDRESS {:}".format(SERVER_ADDRESS))

        # SERVER_ADDRESS = "127.0.0.1:9092"

        ADMIN_CLIENT   = AdminClient({'bootstrap.servers': SERVER_ADDRESS })

        return ADMIN_CLIENT
@@ -55,7 +53,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))

        LOGGER.debug("Kafka server address is: {:} ".format(KafkaConfig.get_kafka_address()))

        if( KafkaTopic.create_new_topic_if_not_exists( all_topics )):

            LOGGER.debug("All topics are created sucsessfully")

            return True
@@ -73,14 +71,14 @@ class KafkaTopic(Enum): 
        LOGGER.debug("Topics names to be verified and created: {:}".format(new_topics))

        for topic in new_topics:

            try:

                topic_metadata = KafkaConfig.ADMIN_CLIENT.value.list_topics(timeout=5)

                topic_metadata = KafkaConfig.get_admin_client().list_topics(timeout=5)

                # LOGGER.debug("Existing topic list: {:}".format(topic_metadata.topics))

                if topic not in topic_metadata.topics:

                    # If the topic does not exist, create a new topic

                    print("Topic {:} does not exist. Creating...".format(topic))

                    LOGGER.debug("Topic {:} does not exist. Creating...".format(topic))

                    new_topic = NewTopic(topic, num_partitions=1, replication_factor=1)

                    KafkaConfig.ADMIN_CLIENT.value.create_topics([new_topic])

                    KafkaConfig.get_admin_client().create_topics([new_topic])

                else:

                    print("Topic name already exists: {:}".format(topic))

                    LOGGER.debug("Topic name already exists: {:}".format(topic))

src/kpi_value_api/service/KpiValueApiServiceServicerImpl.py

+1 −1
Original line number Diff line number Diff line
@@ -43,7 +43,7 @@ class KpiValueApiServiceServicerImpl(KpiValueAPIServiceServicer): 
                       ) -> Empty:

        LOGGER.debug('StoreKpiValues: Received gRPC message object: {:}'.format(request))

        producer_obj = KafkaProducer({

            'bootstrap.servers' : KafkaConfig.get_admin_client()

            'bootstrap.servers' : KafkaConfig.get_kafka_address()

        })

        for kpi_value in request.kpi_value_list:

            kpi_value_to_produce : Tuple [str, Any, Any] = (