prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  About the sessions-subscription-results subscription issue
  ### Environment
  
  - sentry | snuba version ：23.1.1
  
  https://github.com/getsentry/snuba/pull/2737 ，@lynnagara Hello, I have a question about this pr, hope to get your answer, thank you very much
  
  - After removing subscriptions-scheduler-executor-session support in snuba, how to write data to the topic of sessions-subscription-results? Because I see that the crash rate warning code in sentry is still there and has not changed, for example
  
  - https://github.com/getsentry/sentry/pull/28526
  
  https://github.com/getsentry/sentry/blob/8e00dcdf463d916b9ca79ddbe13e99f161d58db1/src/sentry/snuba/query_subscription_consumer.py#L61-L61
  
  My original question is as follows, I have enabled the organizations:incidents function in sentry and subscribed to sessions-results through the following script
  ```bash
  sentry
  run
  query-subscription-consumer
  --topic=sessions-subscription-results
  ```
  Because there is no data in the sessions-subscription-results topic, the crash rate alarm cannot work
  <img width="1568" alt="image" src="https://user-images.githubusercontent.com/18591662/216570393-64748a25-1cd4-4980-966c-f7665dc8482b.png">
  
  
  
  
  
  
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `snuba/cli/subscriptions_scheduler_executor.py`
  Content:
  ```
  1 import signal
  2 from contextlib import contextmanager
  3 from typing import Any, Iterator, Optional, Sequence
  4 
  5 import click
  6 from arroyo import configure_metrics
  7 from arroyo.backends.kafka import KafkaProducer
  8 
  9 from snuba import environment, state
  10 from snuba.attribution.log import flush_attribution_producer
  11 from snuba.datasets.entities.entity_key import EntityKey
  12 from snuba.datasets.entities.factory import get_entity
  13 from snuba.environment import setup_logging, setup_sentry
  14 from snuba.subscriptions.combined_scheduler_executor import (
  15     build_scheduler_executor_consumer,
  16 )
  17 from snuba.utils.metrics.wrapper import MetricsWrapper
  18 from snuba.utils.streams.configuration_builder import build_kafka_producer_configuration
  19 from snuba.utils.streams.metrics_adapter import StreamMetricsAdapter
  20 
  21 
  22 @click.command()
  23 @click.option(
  24     "--dataset",
  25     "dataset_name",
  26     required=True,
  27     type=click.Choice(["events", "transactions", "metrics"]),
  28     help="The dataset to target.",
  29 )
  30 @click.option(
  31     "--entity",
  32     "entity_names",
  33     required=True,
  34     multiple=True,
  35     type=click.Choice(["events", "transactions", "metrics_counters", "metrics_sets"]),
  36     help="The entity to target.",
  37 )
  38 @click.option(
  39     "--consumer-group",
  40     default="snuba-subscription-executor",
  41     help="Consumer group used for consuming the scheduled subscription topic/s.",
  42 )
  43 @click.option(
  44     "--followed-consumer-group",
  45     required=True,
  46     help="Name of the consumer group to follow",
  47 )
  48 @click.option(
  49     "--total-concurrent-queries",
  50     default=64,
  51     type=int,
  52     help="Total max number of concurrent queries for all replicas. Used to calculate max_concurrent_queries.",
  53 )
  54 @click.option(
  55     "--auto-offset-reset",
  56     default="error",
  57     type=click.Choice(["error", "earliest", "latest"]),
  58     help="Kafka consumer auto offset reset.",
  59 )
  60 @click.option(
  61     "--no-strict-offset-reset",
  62     is_flag=True,
  63     help="Forces the kafka consumer auto offset reset.",
  64 )
  65 @click.option("--schedule-ttl", type=int, default=60 * 5)
  66 @click.option("--delay-seconds", type=int)
  67 @click.option(
  68     "--stale-threshold-seconds",
  69     type=int,
  70     help="Skip scheduling if timestamp is beyond this threshold compared to the system time",
  71 )
  72 @click.option("--log-level", help="Logging level to use.")
  73 def subscriptions_scheduler_executor(
  74     *,
  75     dataset_name: str,
  76     entity_names: Sequence[str],
  77     consumer_group: str,
  78     followed_consumer_group: str,
  79     total_concurrent_queries: int,
  80     auto_offset_reset: str,
  81     no_strict_offset_reset: bool,
  82     schedule_ttl: int,
  83     delay_seconds: Optional[int],
  84     stale_threshold_seconds: Optional[int],
  85     log_level: Optional[str],
  86 ) -> None:
  87     """
  88     Combined subscriptions scheduler and executor. Alternative to the separate scheduler and executor processes.
  89     """
  90     setup_logging(log_level)
  91     setup_sentry()
  92 
  93     metrics = MetricsWrapper(
  94         environment.metrics,
  95         "subscriptions.scheduler_executor",
  96         tags={"dataset": dataset_name},
  97     )
  98 
  99     configure_metrics(StreamMetricsAdapter(metrics))
  100 
  101     # Just get the result topic configuration from the first entity. Later we
  102     # check they all have the same result topic anyway before building the consumer.
  103     entity_key = EntityKey(entity_names[0])
  104 
  105     storage = get_entity(entity_key).get_writable_storage()
  106     assert storage is not None
  107     stream_loader = storage.get_table_writer().get_stream_loader()
  108     result_topic_spec = stream_loader.get_subscription_scheduled_topic_spec()
  109     assert result_topic_spec is not None
  110 
  111     producer = KafkaProducer(
  112         build_kafka_producer_configuration(
  113             result_topic_spec.topic,
  114             override_params={"partitioner": "consistent"},
  115         )
  116     )
  117 
  118     processor = build_scheduler_executor_consumer(
  119         dataset_name,
  120         entity_names,
  121         consumer_group,
  122         followed_consumer_group,
  123         producer,
  124         auto_offset_reset,
  125         not no_strict_offset_reset,
  126         schedule_ttl,
  127         delay_seconds,
  128         stale_threshold_seconds,
  129         total_concurrent_queries,
  130         metrics,
  131     )
  132 
  133     def handler(signum: int, frame: Any) -> None:
  134         processor.signal_shutdown()
  135 
  136     signal.signal(signal.SIGINT, handler)
  137     signal.signal(signal.SIGTERM, handler)
  138 
  139     with closing(producer), flush_querylog(), flush_attribution_producer():
  140         processor.run()
  141 
  142 
  143 @contextmanager
  144 def closing(producer: KafkaProducer) -> Iterator[Optional[KafkaProducer]]:
  145     try:
  146         yield producer
  147     finally:
  148         producer.close().result()
  149 
  150 
  151 @contextmanager
  152 def flush_querylog() -> Iterator[None]:
  153     try:
  154         yield
  155     finally:
  156         state.flush_producer()
  157
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/snuba/cli/subscriptions_scheduler_executor.py b/snuba/cli/subscriptions_scheduler_executor.py
  --- a/snuba/cli/subscriptions_scheduler_executor.py
  +++ b/snuba/cli/subscriptions_scheduler_executor.py
  @@ -24,7 +24,7 @@
       "--dataset",
       "dataset_name",
       required=True,
  -    type=click.Choice(["events", "transactions", "metrics"]),
  +    type=click.Choice(["events", "transactions", "metrics", "sessions"]),
       help="The dataset to target.",
   )
   @click.option(
  @@ -32,7 +32,9 @@
       "entity_names",
       required=True,
       multiple=True,
  -    type=click.Choice(["events", "transactions", "metrics_counters", "metrics_sets"]),
  +    type=click.Choice(
  +        ["events", "transactions", "metrics_counters", "metrics_sets", "sessions"]
  +    ),
       help="The entity to target.",
   )
   @click.option(
  
verification_info: {"golden_diff": "diff --git a/snuba/cli/subscriptions_scheduler_executor.py b/snuba/cli/subscriptions_scheduler_executor.py\n--- a/snuba/cli/subscriptions_scheduler_executor.py\n+++ b/snuba/cli/subscriptions_scheduler_executor.py\n@@ -24,7 +24,7 @@\n     \"--dataset\",\n     \"dataset_name\",\n     required=True,\n-    type=click.Choice([\"events\", \"transactions\", \"metrics\"]),\n+    type=click.Choice([\"events\", \"transactions\", \"metrics\", \"sessions\"]),\n     help=\"The dataset to target.\",\n )\n @click.option(\n@@ -32,7 +32,9 @@\n     \"entity_names\",\n     required=True,\n     multiple=True,\n-    type=click.Choice([\"events\", \"transactions\", \"metrics_counters\", \"metrics_sets\"]),\n+    type=click.Choice(\n+        [\"events\", \"transactions\", \"metrics_counters\", \"metrics_sets\", \"sessions\"]\n+    ),\n     help=\"The entity to target.\",\n )\n @click.option(\n", "issue": "About the sessions-subscription-results subscription issue\n### Environment\r\n\r\n- sentry | snuba version \uff1a23.1.1\r\n\r\nhttps://github.com/getsentry/snuba/pull/2737 \uff0c@lynnagara Hello, I have a question about this pr, hope to get your answer, thank you very much\r\n\r\n- After removing subscriptions-scheduler-executor-session support in snuba, how to write data to the topic of sessions-subscription-results? Because I see that the crash rate warning code in sentry is still there and has not changed, for example\r\n\r\n- https://github.com/getsentry/sentry/pull/28526\r\n\r\nhttps://github.com/getsentry/sentry/blob/8e00dcdf463d916b9ca79ddbe13e99f161d58db1/src/sentry/snuba/query_subscription_consumer.py#L61-L61\r\n\r\nMy original question is as follows, I have enabled the organizations:incidents function in sentry and subscribed to sessions-results through the following script\r\n```bash\r\nsentry\r\nrun\r\nquery-subscription-consumer\r\n--topic=sessions-subscription-results\r\n```\r\nBecause there is no data in the sessions-subscription-results topic, the crash rate alarm cannot work\r\n<img width=\"1568\" alt=\"image\" src=\"https://user-images.githubusercontent.com/18591662/216570393-64748a25-1cd4-4980-966c-f7665dc8482b.png\">\r\n\r\n\r\n\r\n\r\n\r\n\n", "before_files": [{"content": "import signal\nfrom contextlib import contextmanager\nfrom typing import Any, Iterator, Optional, Sequence\n\nimport click\nfrom arroyo import configure_metrics\nfrom arroyo.backends.kafka import KafkaProducer\n\nfrom snuba import environment, state\nfrom snuba.attribution.log import flush_attribution_producer\nfrom snuba.datasets.entities.entity_key import EntityKey\nfrom snuba.datasets.entities.factory import get_entity\nfrom snuba.environment import setup_logging, setup_sentry\nfrom snuba.subscriptions.combined_scheduler_executor import (\n    build_scheduler_executor_consumer,\n)\nfrom snuba.utils.metrics.wrapper import MetricsWrapper\nfrom snuba.utils.streams.configuration_builder import build_kafka_producer_configuration\nfrom snuba.utils.streams.metrics_adapter import StreamMetricsAdapter\n\n\n@click.command()\n@click.option(\n    \"--dataset\",\n    \"dataset_name\",\n    required=True,\n    type=click.Choice([\"events\", \"transactions\", \"metrics\"]),\n    help=\"The dataset to target.\",\n)\n@click.option(\n    \"--entity\",\n    \"entity_names\",\n    required=True,\n    multiple=True,\n    type=click.Choice([\"events\", \"transactions\", \"metrics_counters\", \"metrics_sets\"]),\n    help=\"The entity to target.\",\n)\n@click.option(\n    \"--consumer-group\",\n    default=\"snuba-subscription-executor\",\n    help=\"Consumer group used for consuming the scheduled subscription topic/s.\",\n)\n@click.option(\n    \"--followed-consumer-group\",\n    required=True,\n    help=\"Name of the consumer group to follow\",\n)\n@click.option(\n    \"--total-concurrent-queries\",\n    default=64,\n    type=int,\n    help=\"Total max number of concurrent queries for all replicas. Used to calculate max_concurrent_queries.\",\n)\n@click.option(\n    \"--auto-offset-reset\",\n    default=\"error\",\n    type=click.Choice([\"error\", \"earliest\", \"latest\"]),\n    help=\"Kafka consumer auto offset reset.\",\n)\n@click.option(\n    \"--no-strict-offset-reset\",\n    is_flag=True,\n    help=\"Forces the kafka consumer auto offset reset.\",\n)\n@click.option(\"--schedule-ttl\", type=int, default=60 * 5)\n@click.option(\"--delay-seconds\", type=int)\n@click.option(\n    \"--stale-threshold-seconds\",\n    type=int,\n    help=\"Skip scheduling if timestamp is beyond this threshold compared to the system time\",\n)\n@click.option(\"--log-level\", help=\"Logging level to use.\")\ndef subscriptions_scheduler_executor(\n    *,\n    dataset_name: str,\n    entity_names: Sequence[str],\n    consumer_group: str,\n    followed_consumer_group: str,\n    total_concurrent_queries: int,\n    auto_offset_reset: str,\n    no_strict_offset_reset: bool,\n    schedule_ttl: int,\n    delay_seconds: Optional[int],\n    stale_threshold_seconds: Optional[int],\n    log_level: Optional[str],\n) -> None:\n    \"\"\"\n    Combined subscriptions scheduler and executor. Alternative to the separate scheduler and executor processes.\n    \"\"\"\n    setup_logging(log_level)\n    setup_sentry()\n\n    metrics = MetricsWrapper(\n        environment.metrics,\n        \"subscriptions.scheduler_executor\",\n        tags={\"dataset\": dataset_name},\n    )\n\n    configure_metrics(StreamMetricsAdapter(metrics))\n\n    # Just get the result topic configuration from the first entity. Later we\n    # check they all have the same result topic anyway before building the consumer.\n    entity_key = EntityKey(entity_names[0])\n\n    storage = get_entity(entity_key).get_writable_storage()\n    assert storage is not None\n    stream_loader = storage.get_table_writer().get_stream_loader()\n    result_topic_spec = stream_loader.get_subscription_scheduled_topic_spec()\n    assert result_topic_spec is not None\n\n    producer = KafkaProducer(\n        build_kafka_producer_configuration(\n            result_topic_spec.topic,\n            override_params={\"partitioner\": \"consistent\"},\n        )\n    )\n\n    processor = build_scheduler_executor_consumer(\n        dataset_name,\n        entity_names,\n        consumer_group,\n        followed_consumer_group,\n        producer,\n        auto_offset_reset,\n        not no_strict_offset_reset,\n        schedule_ttl,\n        delay_seconds,\n        stale_threshold_seconds,\n        total_concurrent_queries,\n        metrics,\n    )\n\n    def handler(signum: int, frame: Any) -> None:\n        processor.signal_shutdown()\n\n    signal.signal(signal.SIGINT, handler)\n    signal.signal(signal.SIGTERM, handler)\n\n    with closing(producer), flush_querylog(), flush_attribution_producer():\n        processor.run()\n\n\n@contextmanager\ndef closing(producer: KafkaProducer) -> Iterator[Optional[KafkaProducer]]:\n    try:\n        yield producer\n    finally:\n        producer.close().result()\n\n\n@contextmanager\ndef flush_querylog() -> Iterator[None]:\n    try:\n        yield\n    finally:\n        state.flush_producer()\n", "path": "snuba/cli/subscriptions_scheduler_executor.py"}], "after_files": [{"content": "import signal\nfrom contextlib import contextmanager\nfrom typing import Any, Iterator, Optional, Sequence\n\nimport click\nfrom arroyo import configure_metrics\nfrom arroyo.backends.kafka import KafkaProducer\n\nfrom snuba import environment, state\nfrom snuba.attribution.log import flush_attribution_producer\nfrom snuba.datasets.entities.entity_key import EntityKey\nfrom snuba.datasets.entities.factory import get_entity\nfrom snuba.environment import setup_logging, setup_sentry\nfrom snuba.subscriptions.combined_scheduler_executor import (\n    build_scheduler_executor_consumer,\n)\nfrom snuba.utils.metrics.wrapper import MetricsWrapper\nfrom snuba.utils.streams.configuration_builder import build_kafka_producer_configuration\nfrom snuba.utils.streams.metrics_adapter import StreamMetricsAdapter\n\n\n@click.command()\n@click.option(\n    \"--dataset\",\n    \"dataset_name\",\n    required=True,\n    type=click.Choice([\"events\", \"transactions\", \"metrics\", \"sessions\"]),\n    help=\"The dataset to target.\",\n)\n@click.option(\n    \"--entity\",\n    \"entity_names\",\n    required=True,\n    multiple=True,\n    type=click.Choice(\n        [\"events\", \"transactions\", \"metrics_counters\", \"metrics_sets\", \"sessions\"]\n    ),\n    help=\"The entity to target.\",\n)\n@click.option(\n    \"--consumer-group\",\n    default=\"snuba-subscription-executor\",\n    help=\"Consumer group used for consuming the scheduled subscription topic/s.\",\n)\n@click.option(\n    \"--followed-consumer-group\",\n    required=True,\n    help=\"Name of the consumer group to follow\",\n)\n@click.option(\n    \"--total-concurrent-queries\",\n    default=64,\n    type=int,\n    help=\"Total max number of concurrent queries for all replicas. Used to calculate max_concurrent_queries.\",\n)\n@click.option(\n    \"--auto-offset-reset\",\n    default=\"error\",\n    type=click.Choice([\"error\", \"earliest\", \"latest\"]),\n    help=\"Kafka consumer auto offset reset.\",\n)\n@click.option(\n    \"--no-strict-offset-reset\",\n    is_flag=True,\n    help=\"Forces the kafka consumer auto offset reset.\",\n)\n@click.option(\"--schedule-ttl\", type=int, default=60 * 5)\n@click.option(\"--delay-seconds\", type=int)\n@click.option(\n    \"--stale-threshold-seconds\",\n    type=int,\n    help=\"Skip scheduling if timestamp is beyond this threshold compared to the system time\",\n)\n@click.option(\"--log-level\", help=\"Logging level to use.\")\ndef subscriptions_scheduler_executor(\n    *,\n    dataset_name: str,\n    entity_names: Sequence[str],\n    consumer_group: str,\n    followed_consumer_group: str,\n    total_concurrent_queries: int,\n    auto_offset_reset: str,\n    no_strict_offset_reset: bool,\n    schedule_ttl: int,\n    delay_seconds: Optional[int],\n    stale_threshold_seconds: Optional[int],\n    log_level: Optional[str],\n) -> None:\n    \"\"\"\n    Combined subscriptions scheduler and executor. Alternative to the separate scheduler and executor processes.\n    \"\"\"\n    setup_logging(log_level)\n    setup_sentry()\n\n    metrics = MetricsWrapper(\n        environment.metrics,\n        \"subscriptions.scheduler_executor\",\n        tags={\"dataset\": dataset_name},\n    )\n\n    configure_metrics(StreamMetricsAdapter(metrics))\n\n    # Just get the result topic configuration from the first entity. Later we\n    # check they all have the same result topic anyway before building the consumer.\n    entity_key = EntityKey(entity_names[0])\n\n    storage = get_entity(entity_key).get_writable_storage()\n    assert storage is not None\n    stream_loader = storage.get_table_writer().get_stream_loader()\n    result_topic_spec = stream_loader.get_subscription_scheduled_topic_spec()\n    assert result_topic_spec is not None\n\n    producer = KafkaProducer(\n        build_kafka_producer_configuration(\n            result_topic_spec.topic,\n            override_params={\"partitioner\": \"consistent\"},\n        )\n    )\n\n    processor = build_scheduler_executor_consumer(\n        dataset_name,\n        entity_names,\n        consumer_group,\n        followed_consumer_group,\n        producer,\n        auto_offset_reset,\n        not no_strict_offset_reset,\n        schedule_ttl,\n        delay_seconds,\n        stale_threshold_seconds,\n        total_concurrent_queries,\n        metrics,\n    )\n\n    def handler(signum: int, frame: Any) -> None:\n        processor.signal_shutdown()\n\n    signal.signal(signal.SIGINT, handler)\n    signal.signal(signal.SIGTERM, handler)\n\n    with closing(producer), flush_querylog(), flush_attribution_producer():\n        processor.run()\n\n\n@contextmanager\ndef closing(producer: KafkaProducer) -> Iterator[Optional[KafkaProducer]]:\n    try:\n        yield producer\n    finally:\n        producer.close().result()\n\n\n@contextmanager\ndef flush_querylog() -> Iterator[None]:\n    try:\n        yield\n    finally:\n        state.flush_producer()\n", "path": "snuba/cli/subscriptions_scheduler_executor.py"}]}