feat(taskbroker): port subscriptions consumers to tasks (#4395)

Author: lvthanh03

docker-compose.yml

@@ -535,11 +535,6 @@ services:
     command: run consumer --no-strict-offset-reset post-process-forwarder-errors --consumer-group post-process-forwarder --synchronize-commit-log-topic=snuba-commit-log --synchronize-commit-group=snuba-consumers --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
     healthcheck:
       <<: *file_healthcheck_defaults
-  subscription-consumer-events:
-    <<: *sentry_defaults
-    command: run consumer events-subscription-results --consumer-group query-subscription-consumer --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
-    healthcheck:
-      <<: *file_healthcheck_defaults
   ##############################################
   ## Feature Complete Sentry Ingest Consumers ##
   ##############################################
@@ -648,34 +643,6 @@ services:
       <<: *file_healthcheck_defaults
     profiles:
       - feature-complete
-  subscription-consumer-transactions:
-    <<: *sentry_defaults
-    command: run consumer transactions-subscription-results --consumer-group query-subscription-consumer --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
-    healthcheck:
-      <<: *file_healthcheck_defaults
-    profiles:
-      - feature-complete
-  subscription-consumer-eap-items:
-    <<: *sentry_defaults
-    command: run consumer subscription-results-eap-items --consumer-group subscription-results-eap-items --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
-    healthcheck:
-      <<: *file_healthcheck_defaults
-    profiles:
-      - feature-complete
-  subscription-consumer-metrics:
-    <<: *sentry_defaults
-    command: run consumer metrics-subscription-results --consumer-group query-subscription-consumer --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
-    healthcheck:
-      <<: *file_healthcheck_defaults
-    profiles:
-      - feature-complete
-  subscription-consumer-generic-metrics:
-    <<: *sentry_defaults
-    command: run consumer generic-metrics-subscription-results --consumer-group query-subscription-consumer --healthcheck-file-path /tmp/health.txt --max-poll-interval-ms ${SENTRY_KAFKA_MAX_POLL_INTERVAL_MS:-300000}
-    healthcheck:
-      <<: *file_healthcheck_defaults
-    profiles:
-      - feature-complete
   sentry-cleanup:
     <<: *sentry_defaults
     image: sentry-cleanup-self-hosted-local

taskbroker/config.yml

@@ -10,6 +10,46 @@ kafka_topics:
     cluster: default
     consumer_group: taskworker
     produce_only: true
+  events-subscription-results:
+    cluster: default
+    consumer_group: query-subscription-consumer
+    raw:
+      namespace: snuba.subscriptions.events.raw
+      application: sentry
+      taskname: sentry.snuba.query_subscriptions.run.process_events_subscription_from_kafka
+      processing_deadline_duration: 60
+  transactions-subscription-results:
+    cluster: default
+    consumer_group: query-subscription-consumer
+    raw:
+      namespace: snuba.subscriptions.transactions.raw
+      application: sentry
+      taskname: sentry.snuba.query_subscriptions.run.process_transactions_subscription_from_kafka
+      processing_deadline_duration: 60
+  metrics-subscription-results:
+    cluster: default
+    consumer_group: query-subscription-consumer
+    raw:
+      namespace: snuba.subscriptions.metrics.raw
+      application: sentry
+      taskname: sentry.snuba.query_subscriptions.run.process_metrics_subscription_from_kafka
+      processing_deadline_duration: 60
+  generic-metrics-subscription-results:
+    cluster: default
+    consumer_group: query-subscription-consumer
+    raw:
+      namespace: snuba.subscriptions.generic_metrics.raw
+      application: sentry
+      taskname: sentry.snuba.query_subscriptions.run.process_generic_metrics_subscription_from_kafka
+      processing_deadline_duration: 60
+  subscription-results-eap-items:
+    cluster: default
+    consumer_group: subscription-results-eap-items
+    raw:
+      namespace: snuba.subscriptions.eap.raw
+      application: sentry
+      taskname: sentry.snuba.query_subscriptions.run.process_eap_subscription_from_kafka
+      processing_deadline_duration: 60
   profiles:
     cluster: default
     consumer_group: ingest-profiles-consumers-0