feat: Enforce DLQs (#68042)

We had a lot of incidents recently due to lack of DLQs. This is already
mandatory for most new consumers -- let's enforce this in ci now.

.github/CODEOWNERS

@@ -13,7 +13,7 @@
 
 ## Snuba
 /src/sentry/eventstream/                                 @getsentry/owners-snuba
-/src/sentry/consumers/                                   @getsentry/owners-snuba
+/src/sentry/consumers/                                   @getsentry/ops @getsentry/owners-snuba
 /src/sentry/post_process_forwarder/                      @getsentry/owners-snuba
 /src/sentry/utils/snuba.py                               @getsentry/owners-snuba @getsentry/performance
 /src/sentry/utils/snql.py                                @getsentry/owners-snuba

tests/sentry/consumers/test_run.py

@@ -17,3 +17,40 @@ def test_all_importable(consumer_def, settings):
 
     topic = defn["topic"]
     assert topic.value in settings.KAFKA_TOPIC_TO_CLUSTER
+
+
+@pytest.mark.parametrize("consumer_def", list(consumers.KAFKA_CONSUMERS.items()))
+def test_dlq(consumer_def) -> None:
+    post_process_forwarders = [
+        "post-process-forwarder-errors",
+        "post-process-forwarder-transactions",
+        "post-process-forwarder-issue-platform",
+    ]
+    subscription_result_consumers = [
+        "events-subscription-results",
+        "transactions-subscription-results",
+        "generic-metrics-subscription-results",
+        "metrics-subscription-results",
+        "sessions-subscription-results",
+    ]
+    consumers_that_should_have_dlq_but_dont = [
+        "process-spans",
+        "detect-performance-issues",
+        "ingest-monitors",
+        "metrics-last-seen-updater",
+        "generic-metrics-last-seen-updater",
+        "billing-metrics-consumer",
+        "ingest-profiles",
+        "ingest-occurrences",
+        "ingest-replay-recordings",
+        "ingest-replay-recordings-buffered",
+    ]
+
+    consumer_name, defn = consumer_def
+
+    if consumer_name not in (
+        post_process_forwarders
+        + subscription_result_consumers
+        + consumers_that_should_have_dlq_but_dont
+    ):
+        assert defn.get("dlq_topic") is not None, f"{consumer_name} consumer is missing DLQ"