feat(post-process-forwarder) Allow different types of post process forwarders (#29225)

In feat(post-process-forwarder) Different errors and transaction forwarders #28954, we created different processing strategies for errors and transactions post process forwarders.
In this PR, we enable post process forwarder to run with one of the following entities (errors, transactions, all)
The all option would be used in scenarios where we don't need separate post process forwarders. Example devserver, single tenant.
There is no production change with this code since the default option for entity is "all" which means the current post process forwarder would handle all messages

Author: nikhars

src/sentry/eventstream/base.py

@@ -120,6 +120,7 @@ def requires_post_process_forwarder(self):
 
     def run_post_process_forwarder(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,

src/sentry/eventstream/kafka/backend.py

@@ -10,7 +10,10 @@
 from sentry.eventstream.kafka.consumer import SynchronizedConsumer
 from sentry.eventstream.kafka.postprocessworker import (
     _CONCURRENCY_OPTION,
+    ErrorsPostProcessForwarderWorker,
+    PostProcessForwarderType,
     PostProcessForwarderWorker,
+    TransactionsPostProcessForwarderWorker,
     _sampled_eventstream_timer,
 )
 from sentry.eventstream.kafka.protocol import (
@@ -151,6 +154,7 @@ def requires_post_process_forwarder(self):
 
     def _build_consumer(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -169,7 +173,16 @@ def _build_consumer(
         )
 
         concurrency = options.get(_CONCURRENCY_OPTION)
-        worker = PostProcessForwarderWorker(concurrency=concurrency)
+        logger.info(f"Starting post process forwrader to consume {entity} messages")
+        if entity == PostProcessForwarderType.TRANSACTIONS:
+            worker = TransactionsPostProcessForwarderWorker(concurrency=concurrency)
+        elif entity == PostProcessForwarderType.ERRORS:
+            worker = ErrorsPostProcessForwarderWorker(concurrency=concurrency)
+        else:
+            # Default implementation which processes both errors and transactions
+            # irrespective of values in the header. This would most likely be the case
+            # for development environments.
+            worker = PostProcessForwarderWorker(concurrency=concurrency)
 
         consumer = BatchingKafkaConsumer(
             topics=self.topic,
@@ -183,6 +196,7 @@ def _build_consumer(
 
     def run_batched_consumer(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -191,6 +205,7 @@ def run_batched_consumer(
         initial_offset_reset="latest",
     ):
         consumer = self._build_consumer(
+            entity,
             consumer_group,
             commit_log_topic,
             synchronize_commit_group,
@@ -404,6 +419,7 @@ def _get_task_kwargs_and_dispatch(self, message) -> None:
 
     def run_post_process_forwarder(
         self,
+        entity,
         consumer_group,
         commit_log_topic,
         synchronize_commit_group,
@@ -416,6 +432,7 @@ def run_post_process_forwarder(
         if settings.SENTRY_POST_PROCESS_FORWARDER_BATCHING:
             logger.info("Starting batching consumer")
             self.run_batched_consumer(
+                entity,
                 consumer_group,
                 commit_log_topic,
                 synchronize_commit_group,

src/sentry/eventstream/kafka/postprocessworker.py

@@ -2,6 +2,7 @@
 import random
 from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
+from enum import Enum
 from typing import Any, Generator, Mapping, Optional, Sequence
 
 from sentry import options
@@ -25,6 +26,12 @@
 _TRANSACTION_FORWARDER_HEADER = "transaction_forwarder"
 
 
+class PostProcessForwarderType(str, Enum):
+    ERRORS = "errors"
+    TRANSACTIONS = "transactions"
+    ALL = "all"
+
+
 @contextmanager
 def _sampled_eventstream_timer(instance: str) -> Generator[None, None, None]:
     record_metric = random.random() < 0.1

src/sentry/runner/commands/devserver.py

@@ -13,6 +13,7 @@
         "sentry",
         "run",
         "post-process-forwarder",
+        "--entity=all",
         "--loglevel=debug",
         "--commit-batch-size=1",
     ],

src/sentry/runner/commands/run.py

@@ -324,6 +324,12 @@ def cron(**options):
     type=click.Choice(["earliest", "latest"]),
     help="Position in the commit log topic to begin reading from when no prior offset has been recorded.",
 )
+@click.option(
+    "--entity",
+    default="all",
+    type=click.Choice(["all", "errors", "transactions"]),
+    help="The type of entity to process (all, errors, transactions).",
+)
 @log_options()
 @configuration
 def post_process_forwarder(**options):
@@ -332,6 +338,7 @@ def post_process_forwarder(**options):
 
     try:
         eventstream.run_post_process_forwarder(
+            entity=options["entity"],
             consumer_group=options["consumer_group"],
             commit_log_topic=options["commit_log_topic"],
             synchronize_commit_group=options["synchronize_commit_group"],

tests/sentry/eventstream/kafka/test_consumer.py

@@ -691,6 +691,7 @@ def test_post_process_forwarder_batch_consumer(self, dispatch_post_process_group
 
         eventstream = KafkaEventStream()
         consumer = eventstream._build_consumer(
+            entity="all",
             consumer_group=consumer_group,
             commit_log_topic=self.commit_log_topic,
             synchronize_commit_group=synchronize_commit_group,