chore(self-hosted/dev): Create kafka topics from sentry-kafka-schemas (#91247)

This uses sentry-kafka-schemas to determine which topics to create. This
avoids duplicate definitions

Author: hubertdeng123

src/sentry/runner/commands/devserver.py

@@ -414,12 +414,13 @@ def devserver(
             """
                 )
 
-            from sentry.conf.types.kafka_definition import Topic
+            from sentry_kafka_schemas import list_topics
+
             from sentry.utils.batching_kafka_consumer import create_topics
-            from sentry.utils.kafka_config import get_topic_definition
+            from sentry.utils.kafka_config import get_topic_definition_from_name
 
-            for topic in Topic:
-                topic_defn = get_topic_definition(topic)
+            for topic in list_topics():
+                topic_defn = get_topic_definition_from_name(topic)
                 create_topics(topic_defn["cluster"], [topic_defn["real_topic_name"]])
 
             if dev_consumer:

src/sentry/runner/commands/upgrade.py

@@ -83,12 +83,13 @@ def _upgrade(
         nodestore.backend.bootstrap()
 
     if create_kafka_topics:
-        from sentry.conf.types.kafka_definition import Topic
+        from sentry_kafka_schemas import list_topics
+
         from sentry.utils.batching_kafka_consumer import create_topics
-        from sentry.utils.kafka_config import get_topic_definition
+        from sentry.utils.kafka_config import get_topic_definition_from_name
 
-        for topic in Topic:
-            topic_defn = get_topic_definition(topic)
+        for topic in list_topics():
+            topic_defn = get_topic_definition_from_name(topic)
             create_topics(topic_defn["cluster"], [topic_defn["real_topic_name"]])
 
     if repair:

src/sentry/utils/kafka_config.py

@@ -102,3 +102,10 @@ def get_topic_definition(topic: Topic) -> TopicDefinition:
         "cluster": settings.KAFKA_TOPIC_TO_CLUSTER[topic.value],
         "real_topic_name": settings.KAFKA_TOPIC_OVERRIDES.get(topic.value, topic.value),
     }
+
+
+def get_topic_definition_from_name(topic_name: str) -> TopicDefinition:
+    return {
+        "cluster": settings.KAFKA_TOPIC_TO_CLUSTER.get(topic_name, "default"),
+        "real_topic_name": settings.KAFKA_TOPIC_OVERRIDES.get(topic_name, topic_name),
+    }