diff --git a/sentry_sdk/integrations/aiohttp.py b/sentry_sdk/integrations/aiohttp.py
index cf66b9db5a..776d9eef0e 100644
--- a/sentry_sdk/integrations/aiohttp.py
+++ b/sentry_sdk/integrations/aiohttp.py
@@ -554,6 +554,7 @@ def aiohttp_processor(
         if request is None:
             return event
 
+        client_options = sentry_sdk.get_client().options
         with capture_internal_exceptions():
             request_info = event.setdefault("request", {})
 
@@ -571,7 +572,16 @@ def aiohttp_processor(
             # Just attach raw data here if it is within bounds, if available.
             # Unfortunately there's no way to get structured data from aiohttp
             # without awaiting on some coroutine.
-            request_info["data"] = get_aiohttp_request_data(request)
+            if has_data_collection_enabled(client_options):
+                if (
+                    "incoming_request"
+                    in client_options["data_collection"]["http_bodies"]
+                ):
+                    request_info["data"] = get_aiohttp_request_data(request)
+            else:
+                # We never gated this prior to data collection, so it should be attached
+                # when data collection is not enabled.
+                request_info["data"] = get_aiohttp_request_data(request)
 
         return event
 
diff --git a/tests/integrations/aiohttp/test_aiohttp.py b/tests/integrations/aiohttp/test_aiohttp.py
index c8b15d4f17..04e7312b80 100644
--- a/tests/integrations/aiohttp/test_aiohttp.py
+++ b/tests/integrations/aiohttp/test_aiohttp.py
@@ -19,6 +19,7 @@
 
 import sentry_sdk
 from sentry_sdk import capture_message, start_transaction
+from sentry_sdk._types import OVER_SIZE_LIMIT_SUBSTITUTE
 from sentry_sdk.consts import SPANDATA
 from sentry_sdk.integrations.aiohttp import (
     AioHttpIntegration,
@@ -131,6 +132,107 @@ async def hello(request):
     assert request["data"] == json.dumps(body)
 
 
+@pytest.mark.parametrize(
+    "data_collection, expect_body",
+    [
+        pytest.param({}, True, id="data_collection_http_bodies_default"),
+        pytest.param(
+            {"http_bodies": ["incoming_request"]},
+            True,
+            id="data_collection_http_bodies_incoming_request",
+        ),
+        pytest.param(
+            {"http_bodies": []}, False, id="data_collection_http_bodies_empty"
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_aiohttp_request_body_data_collection(
+    sentry_init, aiohttp_client, capture_events, data_collection, expect_body
+):
+    sentry_init(
+        integrations=[AioHttpIntegration()],
+        _experiments={"data_collection": data_collection},
+    )
+
+    body = {"some": "value"}
+
+    async def hello(request):
+        await request.json()
+        1 / 0
+
+    app = web.Application()
+    app.router.add_post("/", hello)
+
+    events = capture_events()
+
+    client = await aiohttp_client(app)
+    resp = await client.post("/", json=body)
+    assert resp.status == 500
+
+    (event,) = events
+    request = event["request"]
+
+    if expect_body:
+        assert request["data"] == json.dumps(body)
+    else:
+        assert "data" not in request
+
+
+@pytest.mark.parametrize(
+    "data_collection, expect_annotated",
+    [
+        pytest.param(
+            {"http_bodies": ["incoming_request"]},
+            True,
+            id="data_collection_http_bodies_incoming_request",
+        ),
+        pytest.param(
+            {"http_bodies": []}, False, id="data_collection_http_bodies_empty"
+        ),
+    ],
+)
+@pytest.mark.asyncio
+async def test_aiohttp_oversized_request_body_data_collection(
+    sentry_init, aiohttp_client, capture_events, data_collection, expect_annotated
+):
+    """
+    The gating happens before the size check. When bodies are collected, an
+    oversized body is still reported as removed because of the size limit; when
+    they are not, it is dropped outright with no annotation.
+    """
+    sentry_init(
+        integrations=[AioHttpIntegration()],
+        max_request_body_size="small",
+        _experiments={"data_collection": data_collection},
+    )
+
+    body = "a" * 2000
+
+    async def hello(request):
+        await request.text()
+        1 / 0
+
+    app = web.Application()
+    app.router.add_post("/", hello)
+
+    events = capture_events()
+
+    client = await aiohttp_client(app)
+    resp = await client.post("/", data=body)
+    assert resp.status == 500
+
+    (event,) = events
+    request_meta = event.get("_meta", {}).get("request", {})
+
+    if expect_annotated:
+        assert event["request"]["data"] == OVER_SIZE_LIMIT_SUBSTITUTE
+        assert request_meta["data"] == {"": {"rem": [["!config", "s"]]}}
+    else:
+        assert "data" not in event["request"]
+        assert "data" not in request_meta
+
+
 @pytest.mark.asyncio
 async def test_403_not_captured(sentry_init, aiohttp_client, capture_events):
     sentry_init(integrations=[AioHttpIntegration()])