Revert "Remove poll"

This reverts commit fbd6d94.

Author: muralibasani

src/karapace/core/schema_reader.py

@@ -262,12 +262,26 @@ def run(self) -> None:
                     LOG.exception("[Schema Topic] Failed to create %r, retrying", self.config.topic_name)
                     self._stop_schema_reader.wait(timeout=SCHEMA_TOPIC_CREATION_TIMEOUT_SECONDS)
 
-            # Subscribe after topic creation; confluent-kafka 2.13+ misbehaves otherwise.
-            # The first consume() in the main loop drives partition assignment.
+            # Subscribe to topic AFTER it has been created
+            # This is critical with confluent-kafka 2.13+ where subscribing to a non-existent topic
+            # causes the consumer to not pick up messages even after the topic is created
             if schema_topic_exists:
                 LOG.debug("[Consumer] Subscribing to topic %r after creation", self.config.topic_name)
                 self.consumer.subscribe([self.config.topic_name])
 
+                # Wait for partition assignment to complete by doing an initial poll
+                # The subscribe() call is asynchronous and partitions are assigned during poll/consume
+                LOG.debug("[Consumer] Waiting for partition assignment")
+                for _ in range(10):  # Try up to 10 times with 100ms timeout each = 1 second total
+                    assignment = self.consumer.assignment()
+                    if assignment:
+                        LOG.debug("[Consumer] Partitions assigned: %s", assignment)
+                        break
+                    # Trigger partition assignment by polling
+                    self.consumer.poll(0.1)
+                else:
+                    LOG.warning("[Consumer] No partitions assigned after waiting, will retry during consumption")
+
             while not self._stop_schema_reader.is_set():
                 if self.offset == OFFSET_UNINITIALIZED:
                     # Handles also a unusual case of purged schemas topic where starting offset can be > 0