diff --git a/src/sentry/monitors/logic/incident_occurrence.py b/src/sentry/monitors/logic/incident_occurrence.py
index d35bf5c72b3ef8..c38bf5f8f4c2d6 100644
--- a/src/sentry/monitors/logic/incident_occurrence.py
+++ b/src/sentry/monitors/logic/incident_occurrence.py
@@ -11,6 +11,7 @@
 from arroyo.backends.kafka import KafkaPayload
 from django.utils.text import get_text_list
 from django.utils.translation import gettext_lazy as _
+from rest_framework import serializers
 from sentry_kafka_schemas.codecs import Codec
 from sentry_kafka_schemas.schema_types.monitors_incident_occurrences_v1 import IncidentOccurrence
 
@@ -28,6 +29,7 @@
     MonitorIncident,
 )
 from sentry.monitors.utils import get_detector_for_monitor
+from sentry.types.actor import validate_actor
 from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer
 from sentry.utils.kafka_config import get_topic_definition
 
@@ -149,6 +151,15 @@ def send_incident_occurrence(
     if detector:
         evidence_data["detector_id"] = detector.id
 
+    owner_actor = monitor_env.monitor.owner_actor
+    if owner_actor:
+        try:
+            validate_actor(owner_actor, monitor_env.monitor.organization_id)
+        except serializers.ValidationError:
+            # If the owner is no longer valid, unassign it from the monitor
+            owner_actor = None
+            monitor_env.monitor.update(owner_user_id=None, owner_team_id=None)
+
     occurrence = IssueOccurrence(
         id=uuid.uuid4().hex,
         resource_id=None,
@@ -179,7 +190,7 @@ def send_incident_occurrence(
         culprit="",
         detection_time=failed_checkin.date_added,
         level="error",
-        assignee=monitor_env.monitor.owner_actor,
+        assignee=owner_actor,
     )
 
     if failed_checkin.trace_id:
diff --git a/src/sentry/types/actor.py b/src/sentry/types/actor.py
index a8692cf6c06ede..80ae20f22c853a 100644
--- a/src/sentry/types/actor.py
+++ b/src/sentry/types/actor.py
@@ -297,8 +297,6 @@ def owner(self, actor: Actor | None) -> None: ...
 
 
 def parse_and_validate_actor(actor_identifier: str | None, organization_id: int) -> Actor | None:
-    from sentry.models.organizationmember import OrganizationMember
-    from sentry.models.team import Team
 
     if not actor_identifier:
         return None
@@ -309,6 +307,15 @@ def parse_and_validate_actor(actor_identifier: str | None, organization_id: int)
         raise serializers.ValidationError(
             "Could not parse actor. Format should be `type:id` where type is `team` or `user`."
         )
+
+    validate_actor(actor, organization_id)
+    return actor
+
+
+def validate_actor(actor: Actor, organization_id: int) -> None:
+    from sentry.models.organizationmember import OrganizationMember
+    from sentry.models.team import Team
+
     try:
         obj = actor.resolve()
     except Actor.InvalidActor:
@@ -322,5 +329,3 @@ def parse_and_validate_actor(actor_identifier: str | None, organization_id: int)
             organization_id=organization_id, user_id=obj.id
         ).exists():
             raise serializers.ValidationError("User is not a member of this organization")
-
-    return actor
diff --git a/tests/sentry/monitors/logic/test_incident_occurrence.py b/tests/sentry/monitors/logic/test_incident_occurrence.py
index 6a894b45c905b9..25c58273d06732 100644
--- a/tests/sentry/monitors/logic/test_incident_occurrence.py
+++ b/tests/sentry/monitors/logic/test_incident_occurrence.py
@@ -345,6 +345,32 @@ def test_queue_incident_occurrence(self, mock_producer: mock.MagicMock) -> None:
             Topic("monitors-test-topic"), test_payload
         )
 
+    @mock.patch("sentry.monitors.logic.incident_occurrence.produce_occurrence_to_kafka")
+    def test_send_incident_occurrence_invalid_owner(
+        self, mock_produce_occurrence_to_kafka: mock.MagicMock
+    ) -> None:
+        self.build_occurrence_test_data()
+        team = self.create_team(organization=self.organization)
+        self.monitor.update(owner_team_id=team.id)
+        self.monitor.refresh_from_db()
+        other_org = self.create_organization()
+        team.update(organization_id=other_org.id)
+
+        send_incident_occurrence(
+            self.failed_checkin,
+            [self.timeout_checkin, self.failed_checkin],
+            self.incident,
+            self.last_checkin,
+        )
+
+        self.monitor.refresh_from_db()
+        assert self.monitor.owner_team_id is None
+        assert self.monitor.owner_user_id is None
+        assert mock_produce_occurrence_to_kafka.call_count == 1
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
+        assert occurrence.assignee is None
+
     @mock.patch("sentry.monitors.logic.incident_occurrence.send_incident_occurrence")
     @mock.patch("sentry.monitors.logic.incident_occurrence.queue_incident_occurrence")
     def test_dispatch_incident_occurrence(