feat(consumer): Make max poll time configurable (#51020)

Adds `max-poll-interval-ms` to the unified consumer CLI. We may want to
pass in max poll time < 5 min to see if it helps mitigate INC-402

Author: ayirr7

src/sentry/consumers/__init__.py

@@ -202,6 +202,7 @@ def get_stream_processor(
     auto_offset_reset: str,
     strict_offset_reset: bool,
     join_timeout: Optional[float],
+    max_poll_interval_ms: Optional[int] = None,
     **options,
 ) -> StreamProcessor:
     try:
@@ -254,6 +255,9 @@ def get_stream_processor(
         strict_offset_reset=strict_offset_reset,
     )
 
+    if max_poll_interval_ms is not None:
+        consumer_config["max.poll.interval.ms"] = max_poll_interval_ms
+
     consumer = KafkaConsumer(consumer_config)
 
     return StreamProcessor(

src/sentry/runner/commands/run.py

@@ -677,6 +677,10 @@ def profiles_consumer(**options):
     help="Position in the commit log topic to begin reading from when no prior offset has been recorded.",
 )
 @click.option("--join-timeout", type=float, help="Join timeout in seconds.", default=None)
+@click.option(
+    "--max-poll-interval-ms",
+    type=int,
+)
 @strict_offset_reset_option()
 @configuration
 def basic_consumer(consumer_name, consumer_args, topic, **options):