diff --git a/sentry_sdk/integrations/rq.py b/sentry_sdk/integrations/rq.py
index 4a2748113b..d6b51ba77d 100644
--- a/sentry_sdk/integrations/rq.py
+++ b/sentry_sdk/integrations/rq.py
@@ -15,6 +15,7 @@
     capture_internal_exceptions,
     event_from_exception,
     format_timestamp,
+    has_data_collection_enabled,
     parse_version,
 )
 
@@ -186,19 +187,21 @@ def event_processor(event: "Event", hint: "dict[str, Any]") -> "Event":
                 rq_job = {
                     "job_id": job.id,
                     "func": job.func_name,
-                    "args": (
-                        job.args
-                        if should_send_default_pii()
-                        else SENSITIVE_DATA_SUBSTITUTE
-                    ),
-                    "kwargs": (
-                        job.kwargs
-                        if should_send_default_pii()
-                        else SENSITIVE_DATA_SUBSTITUTE
-                    ),
                     "description": job.description,
                 }
 
+                client_options = sentry_sdk.get_client().options
+                if has_data_collection_enabled(client_options):
+                    if client_options["data_collection"]["queues"]:
+                        rq_job["args"] = job.args
+                        rq_job["kwargs"] = job.kwargs
+                elif should_send_default_pii():
+                    rq_job["args"] = job.args
+                    rq_job["kwargs"] = job.kwargs
+                else:
+                    rq_job["args"] = SENSITIVE_DATA_SUBSTITUTE
+                    rq_job["kwargs"] = SENSITIVE_DATA_SUBSTITUTE
+
                 if job.enqueued_at:
                     rq_job["enqueued_at"] = format_timestamp(job.enqueued_at)
                 if job.started_at:
diff --git a/tests/integrations/rq/test_rq.py b/tests/integrations/rq/test_rq.py
index e243a8f2f3..1e4013f4ea 100644
--- a/tests/integrations/rq/test_rq.py
+++ b/tests/integrations/rq/test_rq.py
@@ -35,7 +35,7 @@ def _patch_rq_get_server_version(monkeypatch):
                 pass
 
 
-def crashing_job(foo):
+def crashing_job(foo, b=1):
     1 / 0
 
 
@@ -108,6 +108,97 @@ def test_basic(
         assert "started_at" in extra
 
 
+@pytest.mark.parametrize(
+    "init_kwargs,expected_args,expected_kwargs",
+    [
+        pytest.param(
+            {"_experiments": {"data_collection": {}}},
+            [1],
+            {"b": 0},
+            id="data_collection_default",
+        ),
+        pytest.param(
+            {"_experiments": {"data_collection": {"queues": True}}},
+            [1],
+            {"b": 0},
+            id="data_collection_queues_on",
+        ),
+        pytest.param(
+            {"_experiments": {"data_collection": {"queues": False}}},
+            None,
+            None,
+            id="data_collection_queues_off",
+        ),
+        pytest.param(
+            {"send_default_pii": False},
+            SENSITIVE_DATA_SUBSTITUTE,
+            SENSITIVE_DATA_SUBSTITUTE,
+            id="no_pii",
+        ),
+        pytest.param(
+            {
+                "_experiments": {"data_collection": {"queues": False}},
+                "send_default_pii": False,
+            },
+            None,
+            None,
+            id="data_collection_queues_off_with_no_pii",
+        ),
+        pytest.param(
+            {
+                "_experiments": {"data_collection": {"queues": True}},
+                "send_default_pii": False,
+            },
+            [1],
+            {"b": 0},
+            id="data_collection_queues_on_with_no_pii",
+        ),
+    ],
+)
+@pytest.mark.parametrize("span_streaming", [True, False])
+def test_job_args_kwargs_data_collection(
+    sentry_init,
+    capture_events,
+    capture_items,
+    span_streaming,
+    init_kwargs,
+    expected_args,
+    expected_kwargs,
+):
+    sentry_init(
+        integrations=[RqIntegration()],
+        trace_lifecycle="stream" if span_streaming else "static",
+        **init_kwargs,
+    )
+
+    queue = rq.Queue(connection=FakeStrictRedis())
+    worker = rq.SimpleWorker([queue], connection=queue.connection)
+
+    if span_streaming:
+        items = capture_items("event")
+
+        queue.enqueue(crashing_job, 1, b=0)
+        worker.work(burst=True)
+
+        (event,) = (item.payload for item in items)
+    else:
+        events = capture_events()
+
+        queue.enqueue(crashing_job, 1, b=0)
+        worker.work(burst=True)
+
+        (event,) = events
+
+    rq_job = event["extra"]["rq-job"]
+
+    if expected_args is None:
+        assert "args" not in rq_job
+        assert "kwargs" not in rq_job
+    else:
+        assert rq_job["args"] == expected_args
+        assert rq_job["kwargs"] == expected_kwargs
+
+
 @pytest.mark.parametrize("span_streaming", [True, False])
 def test_transport_shutdown(
     sentry_init, capture_events_forksafe, capture_items_forksafe, span_streaming