prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Streamline issue platfrom message processing for non-Kafka envs
  From https://github.com/getsentry/sentry/pull/59330#pullrequestreview-1713484895, 
  
  We can simplify the logic and make our tests more meaningful by not duplicating the message processing logic in dev environments. Instead, we can massage the message format to match a Kafka payloads and directly call `process_message`. 
  
  </issue>
  <code>
  [start of src/sentry/issues/producer.py]
  1 from __future__ import annotations
  2 
  3 import logging
  4 from typing import Any, Dict, MutableMapping, Optional, cast
  5 
  6 from arroyo import Topic
  7 from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
  8 from django.conf import settings
  9 
  10 from sentry import features
  11 from sentry.issues.issue_occurrence import IssueOccurrence
  12 from sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints, update_status
  13 from sentry.issues.status_change_message import StatusChangeMessage
  14 from sentry.models.project import Project
  15 from sentry.services.hybrid_cloud import ValueEqualityEnum
  16 from sentry.utils import json
  17 from sentry.utils.arroyo_producer import SingletonProducer
  18 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
  19 
  20 logger = logging.getLogger(__name__)
  21 
  22 
  23 class PayloadType(ValueEqualityEnum):
  24     OCCURRENCE = "occurrence"
  25     STATUS_CHANGE = "status_change"
  26 
  27 
  28 def _get_occurrence_producer() -> KafkaProducer:
  29     cluster_name = get_topic_definition(settings.KAFKA_INGEST_OCCURRENCES)["cluster"]
  30     producer_config = get_kafka_producer_cluster_options(cluster_name)
  31     producer_config.pop("compression.type", None)
  32     producer_config.pop("message.max.bytes", None)
  33     return KafkaProducer(build_kafka_configuration(default_config=producer_config))
  34 
  35 
  36 _occurrence_producer = SingletonProducer(
  37     _get_occurrence_producer, max_futures=settings.SENTRY_ISSUE_PLATFORM_FUTURES_MAX_LIMIT
  38 )
  39 
  40 
  41 def produce_occurrence_to_kafka(
  42     payload_type: PayloadType | None = PayloadType.OCCURRENCE,
  43     occurrence: IssueOccurrence | None = None,
  44     status_change: StatusChangeMessage | None = None,
  45     event_data: Optional[Dict[str, Any]] = None,
  46 ) -> None:
  47     payload_data = None
  48     if payload_type == PayloadType.OCCURRENCE:
  49         payload_data = _prepare_occurrence_message(occurrence, event_data)
  50     elif payload_type == PayloadType.STATUS_CHANGE:
  51         payload_data = _prepare_status_change_message(status_change)
  52     else:
  53         raise NotImplementedError(f"Unknown payload type: {payload_type}")
  54 
  55     if payload_data is None:
  56         return
  57 
  58     payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
  59     _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
  60 
  61 
  62 def _prepare_occurrence_message(
  63     occurrence: IssueOccurrence | None, event_data: Optional[Dict[str, Any]]
  64 ) -> MutableMapping[str, Any] | None:
  65     if not occurrence:
  66         raise ValueError("occurrence must be provided")
  67     if event_data and occurrence.event_id != event_data["event_id"]:
  68         raise ValueError("Event id on occurrence and event_data must be the same")
  69     if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
  70         # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
  71         # write to the issue platform directly
  72         from sentry.issues.ingest import process_occurrence_data
  73         from sentry.issues.occurrence_consumer import (
  74             lookup_event_and_process_issue_occurrence,
  75             process_event_and_issue_occurrence,
  76         )
  77 
  78         occurrence_dict = occurrence.to_dict()
  79         process_occurrence_data(occurrence_dict)
  80         if event_data:
  81             process_event_and_issue_occurrence(occurrence_dict, event_data)
  82         else:
  83             lookup_event_and_process_issue_occurrence(occurrence_dict)
  84         return None
  85 
  86     payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())
  87     payload_data["payload_type"] = PayloadType.OCCURRENCE.value
  88     if event_data:
  89         payload_data["event"] = event_data
  90 
  91     return payload_data
  92 
  93 
  94 def _prepare_status_change_message(
  95     status_change: StatusChangeMessage | None,
  96 ) -> MutableMapping[str, Any] | None:
  97     if not status_change:
  98         raise ValueError("status_change must be provided")
  99 
  100     organization = Project.objects.get(id=status_change.project_id).organization
  101     if not features.has("organizations:issue-platform-api-crons-sd", organization):
  102         return None
  103 
  104     if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
  105         # Do the change
  106         # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
  107         # write to the issue platform directly
  108         from sentry.issues.ingest import process_occurrence_data
  109 
  110         process_occurrence_data(status_change.to_dict())
  111         fingerprint = status_change.fingerprint
  112         groups_by_fingerprints = bulk_get_groups_from_fingerprints(
  113             [(status_change.project_id, fingerprint)]
  114         )
  115 
  116         key = (status_change.project_id, fingerprint[0])
  117         group = groups_by_fingerprints.get(key, None)
  118         if not group:
  119             return None
  120         update_status(group, status_change.to_dict())
  121         return None
  122 
  123     payload_data = cast(MutableMapping[str, Any], status_change.to_dict())
  124     payload_data["payload_type"] = PayloadType.STATUS_CHANGE.value
  125     return payload_data
  126
  [end of src/sentry/issues/producer.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py
  --- a/src/sentry/issues/producer.py
  +++ b/src/sentry/issues/producer.py
  @@ -5,11 +5,12 @@
   
   from arroyo import Topic
   from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
  +from arroyo.types import Message, Value
   from django.conf import settings
   
   from sentry import features
   from sentry.issues.issue_occurrence import IssueOccurrence
  -from sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints, update_status
  +from sentry.issues.run import process_message
   from sentry.issues.status_change_message import StatusChangeMessage
   from sentry.models.project import Project
   from sentry.services.hybrid_cloud import ValueEqualityEnum
  @@ -56,6 +57,12 @@
           return
   
       payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
  +    if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
  +        # If we're not running Kafka then we're just in dev.
  +        # Skip producing to Kafka and just process the message directly
  +        process_message(Message(Value(payload=payload, committable={})))
  +        return
  +
       _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
   
   
  @@ -66,22 +73,6 @@
           raise ValueError("occurrence must be provided")
       if event_data and occurrence.event_id != event_data["event_id"]:
           raise ValueError("Event id on occurrence and event_data must be the same")
  -    if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
  -        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
  -        # write to the issue platform directly
  -        from sentry.issues.ingest import process_occurrence_data
  -        from sentry.issues.occurrence_consumer import (
  -            lookup_event_and_process_issue_occurrence,
  -            process_event_and_issue_occurrence,
  -        )
  -
  -        occurrence_dict = occurrence.to_dict()
  -        process_occurrence_data(occurrence_dict)
  -        if event_data:
  -            process_event_and_issue_occurrence(occurrence_dict, event_data)
  -        else:
  -            lookup_event_and_process_issue_occurrence(occurrence_dict)
  -        return None
   
       payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())
       payload_data["payload_type"] = PayloadType.OCCURRENCE.value
  @@ -101,25 +92,6 @@
       if not features.has("organizations:issue-platform-api-crons-sd", organization):
           return None
   
  -    if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
  -        # Do the change
  -        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
  -        # write to the issue platform directly
  -        from sentry.issues.ingest import process_occurrence_data
  -
  -        process_occurrence_data(status_change.to_dict())
  -        fingerprint = status_change.fingerprint
  -        groups_by_fingerprints = bulk_get_groups_from_fingerprints(
  -            [(status_change.project_id, fingerprint)]
  -        )
  -
  -        key = (status_change.project_id, fingerprint[0])
  -        group = groups_by_fingerprints.get(key, None)
  -        if not group:
  -            return None
  -        update_status(group, status_change.to_dict())
  -        return None
  -
       payload_data = cast(MutableMapping[str, Any], status_change.to_dict())
       payload_data["payload_type"] = PayloadType.STATUS_CHANGE.value
       return payload_data
  
verification_info: {"golden_diff": "diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py\n--- a/src/sentry/issues/producer.py\n+++ b/src/sentry/issues/producer.py\n@@ -5,11 +5,12 @@\n \n from arroyo import Topic\n from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration\n+from arroyo.types import Message, Value\n from django.conf import settings\n \n from sentry import features\n from sentry.issues.issue_occurrence import IssueOccurrence\n-from sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints, update_status\n+from sentry.issues.run import process_message\n from sentry.issues.status_change_message import StatusChangeMessage\n from sentry.models.project import Project\n from sentry.services.hybrid_cloud import ValueEqualityEnum\n@@ -56,6 +57,12 @@\n         return\n \n     payload = KafkaPayload(None, json.dumps(payload_data).encode(\"utf-8\"), [])\n+    if settings.SENTRY_EVENTSTREAM != \"sentry.eventstream.kafka.KafkaEventStream\":\n+        # If we're not running Kafka then we're just in dev.\n+        # Skip producing to Kafka and just process the message directly\n+        process_message(Message(Value(payload=payload, committable={})))\n+        return\n+\n     _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)\n \n \n@@ -66,22 +73,6 @@\n         raise ValueError(\"occurrence must be provided\")\n     if event_data and occurrence.event_id != event_data[\"event_id\"]:\n         raise ValueError(\"Event id on occurrence and event_data must be the same\")\n-    if settings.SENTRY_EVENTSTREAM != \"sentry.eventstream.kafka.KafkaEventStream\":\n-        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just\n-        # write to the issue platform directly\n-        from sentry.issues.ingest import process_occurrence_data\n-        from sentry.issues.occurrence_consumer import (\n-            lookup_event_and_process_issue_occurrence,\n-            process_event_and_issue_occurrence,\n-        )\n-\n-        occurrence_dict = occurrence.to_dict()\n-        process_occurrence_data(occurrence_dict)\n-        if event_data:\n-            process_event_and_issue_occurrence(occurrence_dict, event_data)\n-        else:\n-            lookup_event_and_process_issue_occurrence(occurrence_dict)\n-        return None\n \n     payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())\n     payload_data[\"payload_type\"] = PayloadType.OCCURRENCE.value\n@@ -101,25 +92,6 @@\n     if not features.has(\"organizations:issue-platform-api-crons-sd\", organization):\n         return None\n \n-    if settings.SENTRY_EVENTSTREAM != \"sentry.eventstream.kafka.KafkaEventStream\":\n-        # Do the change\n-        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just\n-        # write to the issue platform directly\n-        from sentry.issues.ingest import process_occurrence_data\n-\n-        process_occurrence_data(status_change.to_dict())\n-        fingerprint = status_change.fingerprint\n-        groups_by_fingerprints = bulk_get_groups_from_fingerprints(\n-            [(status_change.project_id, fingerprint)]\n-        )\n-\n-        key = (status_change.project_id, fingerprint[0])\n-        group = groups_by_fingerprints.get(key, None)\n-        if not group:\n-            return None\n-        update_status(group, status_change.to_dict())\n-        return None\n-\n     payload_data = cast(MutableMapping[str, Any], status_change.to_dict())\n     payload_data[\"payload_type\"] = PayloadType.STATUS_CHANGE.value\n     return payload_data\n", "issue": "Streamline issue platfrom message processing for non-Kafka envs\nFrom https://github.com/getsentry/sentry/pull/59330#pullrequestreview-1713484895, \n\nWe can simplify the logic and make our tests more meaningful by not duplicating the message processing logic in dev environments. Instead, we can massage the message format to match a Kafka payloads and directly call `process_message`. \n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nfrom typing import Any, Dict, MutableMapping, Optional, cast\n\nfrom arroyo import Topic\nfrom arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration\nfrom django.conf import settings\n\nfrom sentry import features\nfrom sentry.issues.issue_occurrence import IssueOccurrence\nfrom sentry.issues.status_change_consumer import bulk_get_groups_from_fingerprints, update_status\nfrom sentry.issues.status_change_message import StatusChangeMessage\nfrom sentry.models.project import Project\nfrom sentry.services.hybrid_cloud import ValueEqualityEnum\nfrom sentry.utils import json\nfrom sentry.utils.arroyo_producer import SingletonProducer\nfrom sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition\n\nlogger = logging.getLogger(__name__)\n\n\nclass PayloadType(ValueEqualityEnum):\n    OCCURRENCE = \"occurrence\"\n    STATUS_CHANGE = \"status_change\"\n\n\ndef _get_occurrence_producer() -> KafkaProducer:\n    cluster_name = get_topic_definition(settings.KAFKA_INGEST_OCCURRENCES)[\"cluster\"]\n    producer_config = get_kafka_producer_cluster_options(cluster_name)\n    producer_config.pop(\"compression.type\", None)\n    producer_config.pop(\"message.max.bytes\", None)\n    return KafkaProducer(build_kafka_configuration(default_config=producer_config))\n\n\n_occurrence_producer = SingletonProducer(\n    _get_occurrence_producer, max_futures=settings.SENTRY_ISSUE_PLATFORM_FUTURES_MAX_LIMIT\n)\n\n\ndef produce_occurrence_to_kafka(\n    payload_type: PayloadType | None = PayloadType.OCCURRENCE,\n    occurrence: IssueOccurrence | None = None,\n    status_change: StatusChangeMessage | None = None,\n    event_data: Optional[Dict[str, Any]] = None,\n) -> None:\n    payload_data = None\n    if payload_type == PayloadType.OCCURRENCE:\n        payload_data = _prepare_occurrence_message(occurrence, event_data)\n    elif payload_type == PayloadType.STATUS_CHANGE:\n        payload_data = _prepare_status_change_message(status_change)\n    else:\n        raise NotImplementedError(f\"Unknown payload type: {payload_type}\")\n\n    if payload_data is None:\n        return\n\n    payload = KafkaPayload(None, json.dumps(payload_data).encode(\"utf-8\"), [])\n    _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)\n\n\ndef _prepare_occurrence_message(\n    occurrence: IssueOccurrence | None, event_data: Optional[Dict[str, Any]]\n) -> MutableMapping[str, Any] | None:\n    if not occurrence:\n        raise ValueError(\"occurrence must be provided\")\n    if event_data and occurrence.event_id != event_data[\"event_id\"]:\n        raise ValueError(\"Event id on occurrence and event_data must be the same\")\n    if settings.SENTRY_EVENTSTREAM != \"sentry.eventstream.kafka.KafkaEventStream\":\n        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just\n        # write to the issue platform directly\n        from sentry.issues.ingest import process_occurrence_data\n        from sentry.issues.occurrence_consumer import (\n            lookup_event_and_process_issue_occurrence,\n            process_event_and_issue_occurrence,\n        )\n\n        occurrence_dict = occurrence.to_dict()\n        process_occurrence_data(occurrence_dict)\n        if event_data:\n            process_event_and_issue_occurrence(occurrence_dict, event_data)\n        else:\n            lookup_event_and_process_issue_occurrence(occurrence_dict)\n        return None\n\n    payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())\n    payload_data[\"payload_type\"] = PayloadType.OCCURRENCE.value\n    if event_data:\n        payload_data[\"event\"] = event_data\n\n    return payload_data\n\n\ndef _prepare_status_change_message(\n    status_change: StatusChangeMessage | None,\n) -> MutableMapping[str, Any] | None:\n    if not status_change:\n        raise ValueError(\"status_change must be provided\")\n\n    organization = Project.objects.get(id=status_change.project_id).organization\n    if not features.has(\"organizations:issue-platform-api-crons-sd\", organization):\n        return None\n\n    if settings.SENTRY_EVENTSTREAM != \"sentry.eventstream.kafka.KafkaEventStream\":\n        # Do the change\n        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just\n        # write to the issue platform directly\n        from sentry.issues.ingest import process_occurrence_data\n\n        process_occurrence_data(status_change.to_dict())\n        fingerprint = status_change.fingerprint\n        groups_by_fingerprints = bulk_get_groups_from_fingerprints(\n            [(status_change.project_id, fingerprint)]\n        )\n\n        key = (status_change.project_id, fingerprint[0])\n        group = groups_by_fingerprints.get(key, None)\n        if not group:\n            return None\n        update_status(group, status_change.to_dict())\n        return None\n\n    payload_data = cast(MutableMapping[str, Any], status_change.to_dict())\n    payload_data[\"payload_type\"] = PayloadType.STATUS_CHANGE.value\n    return payload_data\n", "path": "src/sentry/issues/producer.py"}]}