diff --git a/sentry_sdk/integrations/celery/__init__.py b/sentry_sdk/integrations/celery/__init__.py
index 92ea1a32e1..0338a8499e 100644
--- a/sentry_sdk/integrations/celery/__init__.py
+++ b/sentry_sdk/integrations/celery/__init__.py
@@ -15,14 +15,12 @@
 )
 from sentry_sdk.integrations.celery.utils import _now_seconds_since_epoch
 from sentry_sdk.integrations.logging import ignore_logger_for_events
-from sentry_sdk.scope import Scope, should_send_default_pii
+from sentry_sdk.scope import Scope
 from sentry_sdk.traces import BAGGAGE_HEADER_NAME, SegmentNameSource, Span
 from sentry_sdk.tracing_utils import Baggage
 from sentry_sdk.utils import (
-    SENSITIVE_DATA_SUBSTITUTE,
     capture_internal_exceptions,
     event_from_exception,
-    has_data_collection_enabled,
     parse_version,
     reraise,
 )
@@ -140,16 +138,9 @@ def event_processor(event: "Event", hint: "Hint") -> "Optional[Event]":
             celery_job = {"task_name": task.name}
 
             client_options = sentry_sdk.get_client().options
-            if has_data_collection_enabled(client_options):
-                if client_options["data_collection"]["queues"]:
-                    celery_job["args"] = args
-                    celery_job["kwargs"] = kwargs
-            elif should_send_default_pii():
+            if client_options["data_collection"]["queues"]:
                 celery_job["args"] = args
                 celery_job["kwargs"] = kwargs
-            else:
-                celery_job["args"] = SENSITIVE_DATA_SUBSTITUTE
-                celery_job["kwargs"] = SENSITIVE_DATA_SUBSTITUTE
 
             extra["celery-job"] = celery_job
 
diff --git a/tests/integrations/celery/test_celery.py b/tests/integrations/celery/test_celery.py
index 176b770e65..b55186f71a 100644
--- a/tests/integrations/celery/test_celery.py
+++ b/tests/integrations/celery/test_celery.py
@@ -13,7 +13,6 @@
     _wrap_task_run,
 )
 from sentry_sdk.integrations.celery.beat import _get_headers
-from sentry_sdk.utils import SENSITIVE_DATA_SUBSTITUTE
 
 
 @pytest.fixture
@@ -118,16 +117,14 @@ def celery_invocation(request):
     return request.param
 
 
-@pytest.mark.parametrize("send_default_pii", [True, False])
 def test_simple_with_performance(
     capture_items,
     init_celery,
     celery_invocation,
-    send_default_pii,
 ):
     celery = init_celery(
         traces_sample_rate=1.0,
-        send_default_pii=send_default_pii,
+        data_collection={},
     )
 
     @celery.task(name="dummy_task")
@@ -150,16 +147,9 @@ def dummy_task(x, y):
 
     assert error_event["transaction"] == "dummy_task"
     assert "celery_task_id" in error_event["tags"]
-    if send_default_pii:
-        assert error_event["extra"]["celery-job"] == dict(
-            task_name="dummy_task", **expected_context
-        )
-    else:
-        assert error_event["extra"]["celery-job"] == {
-            "task_name": "dummy_task",
-            "args": SENSITIVE_DATA_SUBSTITUTE,
-            "kwargs": SENSITIVE_DATA_SUBSTITUTE,
-        }
+    assert error_event["extra"]["celery-job"] == dict(
+        task_name="dummy_task", **expected_context
+    )
 
     (exception,) = error_event["exception"]["values"]
     assert exception["type"] == "ZeroDivisionError"
@@ -167,13 +157,10 @@ def dummy_task(x, y):
     assert exception["stacktrace"]["frames"][0]["vars"]["foo"] == "42"
 
 
-@pytest.mark.parametrize("send_default_pii", [True, False])
-def test_simple_without_performance(
-    capture_events, init_celery, celery_invocation, send_default_pii
-):
+def test_simple_without_performance(capture_events, init_celery, celery_invocation):
     celery = init_celery(
         traces_sample_rate=None,
-        send_default_pii=send_default_pii,
+        data_collection={},
     )
     events = capture_events()
 
@@ -199,16 +186,9 @@ def dummy_task(x, y):
     )
     assert error_event["transaction"] == "dummy_task"
     assert "celery_task_id" in error_event["tags"]
-    if send_default_pii:
-        assert error_event["extra"]["celery-job"] == dict(
-            task_name="dummy_task", **expected_context
-        )
-    else:
-        assert error_event["extra"]["celery-job"] == {
-            "task_name": "dummy_task",
-            "args": SENSITIVE_DATA_SUBSTITUTE,
-            "kwargs": SENSITIVE_DATA_SUBSTITUTE,
-        }
+    assert error_event["extra"]["celery-job"] == dict(
+        task_name="dummy_task", **expected_context
+    )
 
     (exception,) = error_event["exception"]["values"]
     assert exception["type"] == "ZeroDivisionError"
@@ -217,50 +197,26 @@ def dummy_task(x, y):
 
 
 @pytest.mark.parametrize(
-    "init_kwargs,expected_args,expected_kwargs",
+    "data_collection,expected_args,expected_kwargs",
     [
         pytest.param(
-            {"data_collection": {}},
+            {},
             "included",
             "included",
             id="data_collection_default",
         ),
         pytest.param(
-            {"data_collection": {"queues": True}},
+            {"queues": True},
             "included",
             "included",
             id="data_collection_queues_on",
         ),
         pytest.param(
-            {"data_collection": {"queues": False}},
+            {"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(
-            {
-                "data_collection": {"queues": False},
-                "send_default_pii": False,
-            },
-            None,
-            None,
-            id="data_collection_queues_off_with_no_pii",
-        ),
-        pytest.param(
-            {
-                "data_collection": {"queues": True},
-                "send_default_pii": False,
-            },
-            "included",
-            "included",
-            id="data_collection_queues_on_with_no_pii",
-        ),
     ],
 )
 @pytest.mark.parametrize(
@@ -276,14 +232,11 @@ def test_task_args_kwargs_data_collection(
     invocation_style,
     task_args,
     task_kwargs,
-    init_kwargs,
+    data_collection,
     expected_args,
     expected_kwargs,
 ):
-    init_dict = {"send_default_pii": True, **init_kwargs}
-    celery = init_celery(
-        **init_dict,
-    )
+    celery = init_celery(data_collection=data_collection)
 
     @celery.task(name="dummy_task")
     def dummy_task(x, y):
@@ -307,9 +260,6 @@ def dummy_task(x, y):
     elif expected_args == "included":
         assert celery_job["args"] == task_args
         assert celery_job["kwargs"] == task_kwargs
-    else:
-        assert celery_job["args"] == SENSITIVE_DATA_SUBSTITUTE
-        assert celery_job["kwargs"] == SENSITIVE_DATA_SUBSTITUTE
 
 
 @pytest.mark.parametrize("task_fails", [True, False], ids=["error", "success"])
@@ -321,6 +271,7 @@ def test_transaction_events(
 ):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task(name="dummy_task")
@@ -402,7 +353,7 @@ def dummy_task():
 
 
 def test_simple_no_propagation(capture_events, init_celery):
-    celery = init_celery(propagate_traces=False)
+    celery = init_celery(propagate_traces=False, data_collection={})
     events = capture_events()
 
     @celery.task(name="dummy_task")
@@ -514,9 +465,7 @@ def test_traces_sampler_gets_task_info_in_sampling_context(
     DictionaryContaining,  # noqa:N803
 ):
     traces_sampler = mock.Mock(return_value=1.0)
-    celery = init_celery(
-        traces_sampler=traces_sampler,
-    )
+    celery = init_celery(traces_sampler=traces_sampler, data_collection={})
 
     @celery.task(name="dog_walk")
     def walk_dogs(x, y):
@@ -588,7 +537,7 @@ def dummy_task(self, x, y):
 
 
 def test_baggage_propagation(init_celery):
-    celery = init_celery(traces_sample_rate=1.0, release="abcdef")
+    celery = init_celery(traces_sample_rate=1.0, release="abcdef", data_collection={})
 
     @celery.task(name="dummy_task", bind=True)
     def dummy_task(self, x, y):
@@ -625,6 +574,7 @@ def test_sentry_propagate_traces_override(init_celery):
         propagate_traces=True,
         traces_sample_rate=1.0,
         release="abcdef",
+        data_collection={},
     )
 
     @celery.task(name="dummy_task", bind=True)
@@ -652,6 +602,7 @@ def dummy_task(self, message):
 def test_apply_async_manually_span(sentry_init):
     sentry_init(
         integrations=[CeleryIntegration()],
+        data_collection={},
     )
 
     def dummy_function(*args, **kwargs):
@@ -664,7 +615,7 @@ def dummy_function(*args, **kwargs):
 
 
 def test_apply_async_no_args(init_celery):
-    celery = init_celery()
+    celery = init_celery(data_collection={})
 
     @celery.task
     def example_task():
@@ -688,6 +639,7 @@ def test_messaging_destination_name_default_exchange(
 ):
     celery_app = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
     mock_request.delivery_info = {"routing_key": routing_key, "exchange": ""}
 
@@ -696,7 +648,9 @@ def task(): ...
 
     items = capture_items("span")
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert (
         process_span.payload["attributes"]["messaging.destination.name"] == routing_key
@@ -715,6 +669,7 @@ def test_messaging_destination_name_nondefault_exchange(
     """
     celery_app = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
     mock_request.delivery_info = {"routing_key": "celery", "exchange": "custom"}
 
@@ -722,8 +677,11 @@ def test_messaging_destination_name_nondefault_exchange(
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert "messaging.destination.name" not in process_span.payload["attributes"]
 
@@ -731,14 +689,18 @@ def task(): ...
 def test_messaging_id(init_celery, capture_items):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task
     def example_task(): ...
 
     items = capture_items("span")
+
     example_task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert "messaging.message.id" in process_span.payload["attributes"]
 
@@ -746,14 +708,18 @@ def example_task(): ...
 def test_retry_count_zero(init_celery, capture_items):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task()
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert process_span.payload["attributes"]["messaging.message.retry.count"] == 0
 
@@ -764,14 +730,18 @@ def test_retry_count_nonzero(mock_request, init_celery, capture_items):
 
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task()
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert process_span.payload["attributes"]["messaging.message.retry.count"] == 3
 
@@ -780,6 +750,7 @@ def task(): ...
 def test_messaging_system(system, init_celery, capture_items):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     # Does not need to be a real URL, since we use always eager
@@ -789,8 +760,11 @@ def test_messaging_system(system, init_celery, capture_items):
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert process_span.payload["attributes"]["messaging.system"] == system
 
@@ -807,6 +781,7 @@ def publish(*args, **kwargs):
     sentry_init(
         integrations=[CeleryIntegration()],
         traces_sample_rate=1.0,
+        data_collection={},
     )
     celery = Celery(__name__, broker=f"{system}://example.com")  # noqa: E231
 
@@ -836,14 +811,18 @@ def task(): ...
 def test_receive_latency(init_celery, capture_items):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task()
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, _execution_span = items
     assert "messaging.message.receive.latency" in process_span.payload["attributes"]
     assert process_span.payload["attributes"]["messaging.message.receive.latency"] > 0
@@ -852,6 +831,7 @@ def task(): ...
 def tests_span_origin_consumer(init_celery, capture_items):
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
     celery.conf.broker_url = "redis://example.com"  # noqa: E231
 
@@ -859,8 +839,11 @@ def tests_span_origin_consumer(init_celery, capture_items):
     def task(): ...
 
     items = capture_items("span")
+
     task.apply_async()
+
     sentry_sdk.flush()
+
     process_span, execution_span = items
     assert execution_span.payload["attributes"]["sentry.origin"] == "auto.queue.celery"
     assert process_span.payload["attributes"]["sentry.origin"] == "auto.queue.celery"
@@ -877,6 +860,7 @@ def publish(*args, **kwargs):
     sentry_init(
         integrations=[CeleryIntegration()],
         traces_sample_rate=1.0,
+        data_collection={},
     )
     celery = Celery(__name__, broker="redis://example.com")  # noqa: E231
 
@@ -911,12 +895,14 @@ def test_send_task_wrapped(
     sentry_init(
         integrations=[CeleryIntegration()],
         traces_sample_rate=1.0,
+        data_collection={},
     )
     celery = Celery(__name__, broker="redis://example.com")  # noqa: E231
 
     items = capture_items("span")
     with sentry_sdk.start_span(name="custom parent") as outer_span:
         celery.send_task("very_creative_task_name", args=(1, 2), kwargs={"foo": "bar"})
+
     sentry_sdk.flush()
 
     (call,) = patched_send_task.call_args_list  # We should have exactly one call
@@ -961,6 +947,7 @@ def test_user_custom_headers_accessible_in_task(init_celery):
     """
     celery = init_celery(
         traces_sample_rate=1.0,
+        data_collection={},
     )
 
     @celery.task(name="custom_headers_task", bind=True)
diff --git a/tests/integrations/celery/test_update_celery_task_headers.py b/tests/integrations/celery/test_update_celery_task_headers.py
index 7660b00c58..97d3f6def5 100644
--- a/tests/integrations/celery/test_update_celery_task_headers.py
+++ b/tests/integrations/celery/test_update_celery_task_headers.py
@@ -69,8 +69,8 @@ def test_monitor_beat_tasks_with_headers(monitor_beat_tasks):
         assert "sentry-monitor-start-timestamp-s" not in outgoing_headers["headers"]
 
 
-def test_span_with_transaction(sentry_init):
-    sentry_init(traces_sample_rate=1.0)
+def test_span_with_segment(sentry_init):
+    sentry_init(traces_sample_rate=1.0, data_collection={})
     headers = {}
     monitor_beat_tasks = False
 
@@ -89,8 +89,8 @@ def test_span_with_transaction(sentry_init):
             )
 
 
-def test_span_with_transaction_custom_headers(sentry_init):
-    sentry_init(traces_sample_rate=1.0)
+def test_span_with_segment_custom_headers(sentry_init):
+    sentry_init(traces_sample_rate=1.0, data_collection={})
     headers = {
         "baggage": BAGGAGE_VALUE,
         "sentry-trace": SENTRY_TRACE_VALUE,
@@ -133,7 +133,7 @@ def test_celery_trace_propagation_default(sentry_init, monitor_beat_tasks):
     The Celery integration has its own mechanism to propagate traces:
     https://docs.sentry.io/platforms/python/integrations/celery/#distributed-traces
     """
-    sentry_init()
+    sentry_init(data_collection={})
 
     headers = {}
     span = None
@@ -169,7 +169,7 @@ def test_celery_trace_propagation_traces_sample_rate(
     The Celery integration has its own mechanism to propagate traces:
     https://docs.sentry.io/platforms/python/integrations/celery/#distributed-traces
     """
-    sentry_init(traces_sample_rate=traces_sample_rate)
+    sentry_init(traces_sample_rate=traces_sample_rate, data_collection={})
 
     headers = {}
     span = None