bugfix for topic creation with quix-based external broker (#530)

Confirmed this solved the issue based on testing with @luisquix

Author: tim-quix

quixstreams/platforms/quix/config.py

@@ -371,6 +371,7 @@ def create_topics(
         topics: List[Topic],
         timeout: Optional[float] = None,
         finalize_timeout: Optional[float] = None,
+        prepend: bool = True,
     ):
         """
         Create topics in a Quix cluster.
@@ -379,17 +380,18 @@ def create_topics(
         :param timeout: response timeout (seconds); Default 30
         :param finalize_timeout: topic finalization timeout (seconds); Default 60
         marked as "Ready" (and thus ready to produce to/consume from).
+        :param prepend: whether to prepend workspace_id during creation attempt.
         """
         logger.info("Attempting to create topics...")
         current_topics = {t["id"]: t for t in self.get_topics(timeout=timeout)}
         finalize = set()
         for topic in topics:
-            topic_name = self.prepend_workspace_id(topic.name)
-            exists = self.prepend_workspace_id(topic_name) in current_topics
-            if not exists or current_topics[topic_name]["status"] != "Ready":
+            name = self.prepend_workspace_id(topic.name) if prepend else topic.name
+            exists = name in current_topics
+            if not exists or current_topics[name]["status"] != "Ready":
                 if exists:
                     logger.debug(
-                        f"Topic {self.strip_workspace_id_prefix(topic_name)} exists but does "
+                        f"Topic {self.strip_workspace_id_prefix(name)} exists but does "
                         f"not have 'Ready' status. Added to finalize check."
                     )
                 else:
@@ -398,14 +400,14 @@ def create_topics(
                     # TODO: more robust error handling to better identify issues
                     # See how it's handled in the admin client and maybe consolidate
                     # logic via TopicManager
-                    except HTTPError as e:
+                    except QuixApiRequestFailure as e:
                         # Topic was maybe created by another instance
-                        if "already exists" not in e.response.text:
+                        if "already exists" not in e.error_text:
                             raise
-                finalize.add(topic_name)
+                finalize.add(name)
             else:
                 logger.debug(
-                    f"Topic {self.strip_workspace_id_prefix(topic_name)} exists and is Ready"
+                    f"Topic {self.strip_workspace_id_prefix(name)} exists and is Ready"
                 )
         logger.info(
             "Topic creations acknowledged; waiting for 'Ready' statuses..."

quixstreams/platforms/quix/topic_manager.py

@@ -63,8 +63,10 @@ def _create_topics(
         :param timeout: creation acknowledge timeout (seconds)
         :param create_timeout: topic finalization timeout (seconds)
         """
+        # note: prepending was already performed as needed, so we skip doing it
+        # within `create_topics`
         self._quix_config_builder.create_topics(
-            topics, timeout=timeout, finalize_timeout=create_timeout
+            topics, timeout=timeout, finalize_timeout=create_timeout, prepend=False
         )
 
     def _resolve_topic_name(self, name: str) -> str:

tests/test_quixstreams/test_platforms/test_quix/test_config.py

@@ -596,6 +596,7 @@ def test_create_topics_parallel_create_attempt(self, topic_manager_topic_factory
 
         mock_response = create_autospec(Response)
         mock_response.text = "already exists"
+        mock_response.status_code = 400
 
         timeout = 4.5
         finalize_timeout = 9.5
@@ -605,7 +606,11 @@ def test_create_topics_parallel_create_attempt(self, topic_manager_topic_factory
         )
         stack = ExitStack()
         create_topic = stack.enter_context(patch.object(cfg_builder, "_create_topic"))
-        create_topic.side_effect = HTTPError(response=mock_response)
+        create_topic.side_effect = QuixApiRequestFailure(
+            url="url",
+            status_code=mock_response.status_code,
+            error_text=mock_response.text,
+        )
         get_topics = stack.enter_context(patch.object(cfg_builder, "get_topics"))
         get_topics.return_value = get_topics_return
         finalize = stack.enter_context(patch.object(cfg_builder, "_finalize_create"))