dynamically creates the kafka server address with env variables.

from enum import Enum

from confluent_kafka import KafkaException

from confluent_kafka.admin import AdminClient, NewTopic

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_IP    = "kafka-service.kafka.svc.cluster.local:9092"

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

    server_address = KFK_SERVER_ADDRESS_TEMPLATE.format(KFK_NAMESPACE, KFK_PORT)

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

class KafkaTopic(Enum):

    REQUEST  = 'topic_request'