feat(taskworker): Make process_profile worker support uncompressed payloads (#95692)

The taskworker producer layer now supports zstd compression. This means
that the zlib task level compression is no longer necessary here. In
order to remove, this PR updates the worker code to take an optional
parameter which decides whether or not the payload should be zlib
decompressed. In a subsequent PR, the producer code will be updated to
send uncompressed payloads

Author: enochtangg

src/sentry/profiles/task.py

@@ -101,15 +101,20 @@ def _get_profiles_producer_from_topic(topic: Topic) -> KafkaProducer:
 logger = logging.getLogger(__name__)
 
 
-def decode_payload(encoded: str) -> dict[str, Any]:
-    try:
-        res = msgpack.unpackb(zlib.decompress(b64decode(encoded.encode("utf-8"))), use_list=False)
-        metrics.incr("profiling.profile_metrics.decompress", tags={"status": "ok"})
-        return res
-    except Exception as e:
-        logger.exception("Failed to decompress compressed profile", extra={"error": e})
-        metrics.incr("profiling.profile_metrics.decompress", tags={"status": "err"})
-        raise
+def decode_payload(encoded: str, compressed_profile: bool) -> dict[str, Any]:
+    if compressed_profile:
+        try:
+            res = msgpack.unpackb(
+                zlib.decompress(b64decode(encoded.encode("utf-8"))), use_list=False
+            )
+            metrics.incr("profiling.profile_metrics.decompress", tags={"status": "ok"})
+            return res
+        except Exception as e:
+            logger.exception("Failed to decompress compressed profile", extra={"error": e})
+            metrics.incr("profiling.profile_metrics.decompress", tags={"status": "err"})
+            raise
+    else:
+        return msgpack.unpackb(b64decode(encoded.encode("utf-8")), use_list=False)
 
 
 def encode_payload(message: dict[str, Any]) -> str:
@@ -146,13 +151,14 @@ def process_profile_task(
     profile: Profile | None = None,
     payload: str | None = None,
     sampled: bool = True,
+    compressed_profile: bool = True,
     **kwargs: Any,
 ) -> None:
     if not sampled and not options.get("profiling.profile_metrics.unsampled_profiles.enabled"):
         return
 
     if payload:
-        message_dict = decode_payload(payload)
+        message_dict = decode_payload(payload, compressed_profile)
 
         profile = json.loads(message_dict["payload"], use_rapid_json=True)