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()

```

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(

{"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"}]}