Merge pull request #130 from salute-developers/PNLP-7328_config_kafka_replyTopic

PNLP-7328: map reply_topic_key to kafka_replyTopic before kafka request

Author: SyrexMinus

core/basic_models/actions/client_profile_actions.py

@@ -2,7 +2,8 @@
 
 from core.basic_models.actions.command import Command
 from core.basic_models.actions.string_actions import StringAction
-from core.configs.global_constants import KAFKA
+from core.configs.config_constants import REPLY_TOPIC_KEY
+from core.configs.global_constants import KAFKA, KAFKA_REPLY_TOPIC
 from core.text_preprocessing.base import BaseTextPreprocessingResult
 from core.utils.pickle_copy import pickle_deepcopy
 from scenarios.user.user_model import User
@@ -62,12 +63,14 @@ def __init__(self, items: Dict[str, Any], id: Optional[str] = None):
         })
         settings_kafka_key = config["template_settings"].get("client_profile_kafka_key")
         self.kafka_key: str = self.kafka_key or settings_kafka_key or self.DEFAULT_KAFKA_KEY
-        self.request_data = {
-            "topic_key": "client_info",
-            "kafka_key": self.kafka_key,
-            "kafka_replyTopic":
-                config["template_settings"]["consumer_topic"]
-        }
+        if self.request_data is None:
+            self.request_data = dict()
+        if "topic_key" not in self.request_data:
+            self.request_data["topic_key"] = "client_info"
+        if "kafka_key" not in self.request_data:
+            self.request_data["kafka_key"] = self.kafka_key
+        if REPLY_TOPIC_KEY not in self.request_data and KAFKA_REPLY_TOPIC not in self.request_data:
+            self.request_data[KAFKA_REPLY_TOPIC] = config["template_settings"]["consumer_topic"]
 
     async def run(self, user: User, text_preprocessing_result: BaseTextPreprocessingResult,
                   params: Optional[Dict[str, Union[str, float, int]]] = None) -> Optional[List[Command]]:
@@ -160,14 +163,16 @@ async def run(self, user: User, text_preprocessing_result: BaseTextPreprocessing
                 "projectId": user.settings["template_settings"]["project_id"]
             }
         })
-        settings_kafka_key: Optional[str] = user.settings["template_settings"].get("client_profile_kafka_key")
-        kafka_key: str = self.kafka_key or settings_kafka_key or self.DEFAULT_KAFKA_KEY
-        self.request_data = {
-            "topic_key": "client_info_remember",
-            "kafka_key": kafka_key,
-            "kafka_replyTopic":
-                user.settings["template_settings"]["consumer_topic"]
-        }
+        if self.request_data is None:
+            self.request_data = dict()
+        if "topic_key" not in self.request_data:
+            self.request_data["topic_key"] = "client_info_remember"
+        if "kafka_key" not in self.request_data:
+            settings_kafka_key: Optional[str] = user.settings["template_settings"].get("client_profile_kafka_key")
+            kafka_key: str = self.kafka_key or settings_kafka_key or self.DEFAULT_KAFKA_KEY
+            self.request_data["kafka_key"] = kafka_key
+        if REPLY_TOPIC_KEY not in self.request_data and KAFKA_REPLY_TOPIC not in self.request_data:
+            self.request_data[KAFKA_REPLY_TOPIC] = user.settings["template_settings"]["consumer_topic"]
 
         commands = await super().run(user, text_preprocessing_result, params)
         return commands

core/configs/config_constants.py

@@ -0,0 +1,2 @@
+REPLY_TOPIC = "reply_topic"
+REPLY_TOPIC_KEY = "reply_topic_key"

core/logging/logger_utils.py

@@ -10,6 +10,7 @@
 import core.logging.logger_constants as log_const
 import scenarios.logging.logger_constants as scenarios_log_const
 from core.basic_models.classifiers.basic_classifiers import Classifier
+from core.configs.global_constants import KAFKA_REPLY_TOPIC
 from core.utils.masking_message import masking
 from core.utils.stats_timer import StatsTimer
 
@@ -22,7 +23,7 @@
 
 
 class LoggerMessageCreator:
-    LOGGER_HEADERS = ["kafka_replyTopic", "app_callback_id"]
+    LOGGER_HEADERS = [KAFKA_REPLY_TOPIC, "app_callback_id"]
     ART_NAMES = [
         "channel", "type", "device_channel", "device_channel_version", "device_platform", "group",
         "device_platform_version", "device_platform_client_type", "csa_profile_id", "test_deploy"

core/mq/kafka/kafka_publisher.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import time
-from typing import Union
+from typing import Union, Optional, List, Tuple, Any
 
 from confluent_kafka import Producer
 
@@ -28,7 +28,7 @@ def __init__(self, config):
             conf["logger"] = debug_logger
         self._producer = Producer(**conf)
 
-    def send(self, value: Union[str, bytes], key=None, topic_key=None, headers=None):
+    def send(self, value: Union[str, bytes], key=None, topic_key=None, headers: Optional[List[Tuple[Any, Any]]] = None):
         try:
             topic = self._config["topic"]
             if topic_key is not None:
@@ -47,7 +47,7 @@ def send(self, value: Union[str, bytes], key=None, topic_key=None, headers=None)
             monitoring.got_counter("kafka_producer_exception")
         self._poll()
 
-    def send_to_topic(self, value, key=None, topic=None, headers=None):
+    def send_to_topic(self, value, key=None, topic=None, headers: Optional[List[Tuple[Any, Any]]] = None):
         try:
             if topic is None:
                 params = {

core/request/kafka_request.py

@@ -13,7 +13,7 @@ class KafkaRequest(BaseRequest):
     TOPIC = "topic"
 
     def __init__(self, items, id=None):
-        super(KafkaRequest, self).__init__(items)
+        super().__init__(items)
         items = items or {}
         self.topic_key = items.get(self.TOPIC_KEY)
         self.kafka_key = items.get(self.KAFKA_KEY)

smart_kit/request/kafka_request.py

@@ -1,27 +1,36 @@
 from core.configs.global_constants import CALLBACK_ID_HEADER, KAFKA_REPLY_TOPIC
 from core.request.kafka_request import KafkaRequest
+from core.configs.config_constants import REPLY_TOPIC_KEY, REPLY_TOPIC
+from smart_kit.configs.settings import Settings
 
 
 class SmartKitKafkaRequest(KafkaRequest):
     KAFKA_REPLY_TOPIC = KAFKA_REPLY_TOPIC
     KAFKA_EXTRA_HEADERS = "kafka_extraHeaders"
+    REPLY_TOPIC_KEY = REPLY_TOPIC_KEY
 
     def __init__(self, items, id=None):
-        super(SmartKitKafkaRequest, self).__init__(items)
+        super().__init__(items)
         items = items or {}
         self._callback_id = items.get(self._callback_id_header_name)
         self._kafka_replyTopic = items.get(self.KAFKA_REPLY_TOPIC)
         self._kafka_extraHeaders = items.get(self.KAFKA_EXTRA_HEADERS) or {}
+        # _reply_topic_key has priority over _kafka_replyTopic
+        self._reply_topic_key = items.get(self.REPLY_TOPIC_KEY)
 
     @property
     def _callback_id_header_name(self):
         return CALLBACK_ID_HEADER
 
     def _get_new_headers(self, source_mq_message):
-        headers_dict = dict(super(SmartKitKafkaRequest, self)._get_new_headers(source_mq_message))
+        headers_dict = dict(super()._get_new_headers(source_mq_message))
         if self._callback_id:
             headers_dict[self._callback_id_header_name] = str(self._callback_id).encode()
-        if self._kafka_replyTopic:
+        if self._reply_topic_key:
+            reply_topics = Settings()["template_settings"][REPLY_TOPIC]
+            mapped_reply_topic = reply_topics[self._reply_topic_key]
+            headers_dict[self.KAFKA_REPLY_TOPIC] = str(mapped_reply_topic).encode()
+        elif self._kafka_replyTopic:
             headers_dict[self.KAFKA_REPLY_TOPIC] = str(self._kafka_replyTopic).encode()
         if self._kafka_extraHeaders:
             for k, v in self._kafka_extraHeaders.items():