fix(crons): Fix connection spikes during partition assignment (#70585)

`on_partitions_assigned` is called whenever partitions are assigned to a
consumer. This calls `_create_strategy`, which calls
`create_with_partitions`.

Since we create the `ThreadPoolExecutor` in `create_parallel_worker`,
this means we create a new threadpool whenever we partition assignment
changes. The reason this causes spikes and isn't sustained is that
presumably the previous `ProcessingStrategy` stops being used and ends
up garbage collected, which results in the `ThreadPoolExecutor` being
garbage collected too.

Author: wedamija

src/sentry/monitors/consumers/monitor_consumer.py

@@ -1028,6 +1028,7 @@ def __init__(
     ) -> None:
         if mode == "parallel":
             self.parallel = True
+            self.parallel_executor = ThreadPoolExecutor(max_workers=self.max_workers)
 
         if max_batch_size is not None:
             self.max_batch_size = max_batch_size
@@ -1041,8 +1042,7 @@ def shutdown(self) -> None:
             self.parallel_executor.shutdown()
 
     def create_parallel_worker(self, commit: Commit) -> ProcessingStrategy[KafkaPayload]:
-        self.parallel_executor = ThreadPoolExecutor(max_workers=self.max_workers)
-
+        assert self.parallel_executor is not None
         batch_processor = RunTask(
             function=partial(process_batch, self.parallel_executor),
             next_step=CommitOffsets(commit),