ref(replays): Use isolation scope instead of Hub in recording consumers (#73280)

The Hub API is deprecated as of Sentry Python SDK 2.0.0, so we should
isolation scope forks instead of Hub clones in the recording consumers

ref #65869

src/sentry/replays/consumers/recording.py

@@ -28,7 +28,7 @@ RECORDINGS_CODEC: Codec[ReplayRecording] = get_topic_codec(Topic.INGEST_REPLAYS_
 class MessageContext:
     message: bytes
     transaction: Span
-    current_hub: sentry_sdk.Hub
+    isolation_scope: sentry_sdk.Scope
 
     # The message attribute can cause large log messages to be emitted which can pin the CPU
     # to 100.
@@ -110,8 +110,8 @@ def initialize_threaded_context(message: Message[KafkaPayload]) -> MessageContex
         sampled=random.random()
         < getattr(settings, "SENTRY_REPLAY_RECORDINGS_CONSUMER_APM_SAMPLING", 0),
     )
-    current_hub = sentry_sdk.Hub(sentry_sdk.Hub.current)
-    return MessageContext(message.payload.value, transaction, current_hub)
+    isolation_scope = sentry_sdk.Scope.get_isolation_scope().fork()
+    return MessageContext(message.payload.value, transaction, isolation_scope)
 
 
 def process_message_threaded(message: Message[MessageContext]) -> Any:
@@ -125,7 +125,7 @@ def process_message_threaded(message: Message[MessageContext]) -> Any:
         logger.exception("Could not decode recording message.")
         return None
 
-    ingest_recording(message_dict, context.transaction, context.current_hub)
+    ingest_recording(message_dict, context.transaction, context.isolation_scope)
 
 
 def process_message(message: Message[KafkaPayload]) -> Any:
@@ -136,7 +136,7 @@ def process_message(message: Message[KafkaPayload]) -> Any:
         sampled=random.random()
         < getattr(settings, "SENTRY_REPLAY_RECORDINGS_CONSUMER_APM_SAMPLING", 0),
     )
-    current_hub = sentry_sdk.Hub(sentry_sdk.Hub.current)
+    isolation_scope = sentry_sdk.Scope.get_isolation_scope().fork()
 
     try:
         message_dict: ReplayRecording = RECORDINGS_CODEC.decode(message.payload.value)
@@ -145,4 +145,4 @@ def process_message(message: Message[KafkaPayload]) -> Any:
         logger.exception("Could not decode recording message.")
         return None
 
-    ingest_recording(message_dict, transaction, current_hub)
+    ingest_recording(message_dict, transaction, isolation_scope)

src/sentry/replays/usecases/ingest/__init__.py

@@ -6,8 +6,9 @@ import zlib
 from datetime import datetime, timezone
 from typing import TypedDict, cast
 
+import sentry_sdk.scope
 from sentry_kafka_schemas.schema_types.ingest_replay_recordings_v1 import ReplayRecording
-from sentry_sdk import Hub, set_tag
+from sentry_sdk import Scope, set_tag
 from sentry_sdk.tracing import Span
 
 from sentry.constants import DataCategory
@@ -66,9 +67,11 @@ class RecordingIngestMessage:
 
 
 @metrics.wraps("replays.usecases.ingest.ingest_recording")
-def ingest_recording(message_dict: ReplayRecording, transaction: Span, current_hub: Hub) -> None:
+def ingest_recording(
+    message_dict: ReplayRecording, transaction: Span, isolation_scope: Scope
+) -> None:
     """Ingest non-chunked recording messages."""
-    with current_hub:
+    with sentry_sdk.scope.use_isolation_scope(isolation_scope):
         with transaction.start_child(
             op="replays.usecases.ingest.ingest_recording",
             description="ingest_recording",