create kafka topics on developer update, first time install and normal deploy

Author: rerpha

installation_and_upgrade/ibex_install_utils/install_tasks.py

@@ -254,6 +254,8 @@ def run_instrument_deploy_main(self) -> None:
         self._python_tasks.update_script_definitions()
         self._python_tasks.remove_instrument_script_githooks()
         self._server_tasks.setup_log_rotation()
+        self._system_tasks.update_kafka_topics()
+
 
     def run_instrument_deploy_pre_stop(self) -> None:
         """Upgrade an instrument. Steps to do before ibex is stopped.
@@ -293,6 +295,7 @@ def run_developer_update(self) -> None:
         self._mysql_tasks.install_mysql(force=False)
         self._system_tasks.check_java_installation()
         self._system_tasks.install_or_upgrade_git()
+        self._system_tasks.update_kafka_topics()
 
     def run_vhd_creation(self) -> None:
         """Automated job which creates a set of VHDs containing all IBEX components.

installation_and_upgrade/ibex_install_utils/kafka_utils.py

@@ -1,10 +1,12 @@
 from kafka import KafkaConsumer
 from kafka.admin import KafkaAdminClient, NewTopic
 
-REQUIRED_SUFFIXES = ["_events", "_sampleEnv", "_runInfo", "_epicsForwarderConfig", "_detSpecMap"]
+REQUIRED_SUFFIXES = ["_events", "_sampleEnv", "_runInfo", "_forwarderConfig", "_detSpecMap",
+                     "_forwarderStorage", "_forwarderStatus", "_runLog", "_areaDetector",
+                     "_monitorHistograms"]
 
 
-def get_existing_topics(kafka_broker):
+def _get_existing_topics(kafka_broker):
     return KafkaConsumer(bootstrap_servers=[kafka_broker]).topics()
 
 
@@ -18,7 +20,7 @@ def add_required_topics(kafka_broker, instrument):
     """
     required_topics = set(instrument + suffix for suffix in REQUIRED_SUFFIXES)
     existing_topics = set(
-        filter(lambda topic: topic.startswith(instrument), get_existing_topics(kafka_broker))
+        filter(lambda topic: topic.startswith(instrument), _get_existing_topics(kafka_broker))
     )
 
     if required_topics != existing_topics:

installation_and_upgrade/ibex_install_utils/tasks/system_tasks.py

@@ -189,7 +189,7 @@ def update_kafka_topics(self) -> None:
         """
         Adds the required kafka topics to the cluster.
         """
-        add_required_topics("livedata.isis.cclrc.ac.uk:9092", self._get_instrument_name())
+        add_required_topics("livedata.isis.cclrc.ac.uk:31092", self._get_instrument_name())
 
     @task("Add Nagios checks")
     def add_nagios_checks(self) -> None: