Sentry before_send/transport dotted-path config never resolved to a callable in Task SDK (silently drops all events)

Start in task-sdk/src/airflow/sdk/execution_time/sentry/configured.py, specifically ConfiguredSentry.prepare_to_enrich_errors(), and compare its [sentry] option handling with the 2.11.0 airflow/sentry.py behavior described in the issue. Verify the configured before_send and transport values are resolved through conf.getimport() before sentry_sdk.init() receives them. Done means dotted paths become callable options even when the Sentry section has schema defaults, without dropping events.

Under which category would you file this issue?

Task SDK

Apache Airflow version

3.2.2 (also reproduced, unchanged, on 3.3.0rc2)

What happened and how to reproduce it?

When [sentry] before_send (e.g. AIRFLOW__SENTRY__BEFORE_SEND=<dotted.path>) is configured,
ConfiguredSentry.prepare_to_enrich_errors() in
task-sdk/src/airflow/sdk/execution_time/sentry/configured.py never resolves it into an
actual callable — it gets passed to sentry_sdk.init() as the raw string:

sentry_config_opts: dict[str, Any] = conf.getsection("sentry") or {}
if sentry_config_opts:
    sentry_config_opts.pop("sentry_on")
    old_way_dsn = sentry_config_opts.pop("sentry_dsn", None)
    new_way_dsn = sentry_config_opts.pop("dsn", None)
    dsn = old_way_dsn or new_way_dsn
    ...
else:
    dsn = None
    if before_send := conf.getimport("sentry", "before_send", fallback=None):
        sentry_config_opts["before_send"] = before_send
    if transport := conf.getimport("sentry", "transport", fallback=None):
        sentry_config_opts["transport"] = transport

if dsn:
    sentry_sdk.init(dsn=dsn, integrations=integrations, **sentry_config_opts)

conf.getsection("sentry") always returns a non-empty dict — it starts from the section's
schema defaults (sentry_on, sentry_dsn both have non-None defaults) — so the
if sentry_config_opts: branch is always taken, and the else branch (the only place
before_send/transport get resolved via conf.getimport()) is unreachable dead code.

As a result, sentry_sdk.init(before_send="my.dotted.path", ...) receives a plain string.
At event-capture time, sentry_sdk's Client._prepare_event calls before_send(event, hint),
raising TypeError: 'str' object is not callable. This is swallowed internally by
capture_internal_exceptions() (logged only via the internal sentry_sdk.errors logger) —
the event is silently dropped with no error surfaced and nothing delivered to Sentry.
Same issue applies to transport.

This is a regression from Airflow 2.x's airflow/sentry.py, where the equivalent
conf.getimport(...) calls ran unconditionally:

sentry_config_opts["before_send"] = conf.getimport("sentry", "before_send", fallback=None)
sentry_config_opts["transport"] = conf.getimport("sentry", "transport", fallback=None)

(see 2.11.0's airflow/sentry.py). This logic appears to have been ported into the Task SDK
reimplementation in #57032, but the branching structure changed such that the resolution
path became unreachable.

Steps to reproduce:

AIRFLOW__SENTRY__SENTRY_ON=True
AIRFLOW__SENTRY__SENTRY_DSN=<any DSN>
AIRFLOW__SENTRY__BEFORE_SEND=path.to.a.real.before_send.function

from airflow.sdk.execution_time.sentry.configured import ConfiguredSentry
cs = ConfiguredSentry()
cs.prepare_to_enrich_errors(executor_integration="")
import sentry_sdk
print(type(sentry_sdk.get_client().options.get("before_send")))
# prints <class 'str'> instead of the resolved function

Any subsequent sentry_sdk.capture_exception() / capture_message() call will silently
fail to deliver its event.

What you think should happen instead?

before_send/transport dotted paths configured via [sentry] should always be resolved
into real callables via conf.getimport(), regardless of what else is set in the [sentry]
section — matching 2.x behavior, e.g.:

sentry_config_opts: dict[str, Any] = conf.getsection("sentry") or {}
sentry_config_opts.pop("sentry_on", None)
old_way_dsn = sentry_config_opts.pop("sentry_dsn", None)
new_way_dsn = sentry_config_opts.pop("dsn", None)
dsn = old_way_dsn or new_way_dsn
if before_send := conf.getimport("sentry", "before_send", fallback=None):
    sentry_config_opts["before_send"] = before_send
if transport := conf.getimport("sentry", "transport", fallback=None):
    sentry_config_opts["transport"] = transport

Operating System

N/A — reproduced via the official apache/airflow:3.2.2-python3.12 and
apache/airflow:3.3.0rc2-python3.12 images.

Deployment

Other Docker-based deployment

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Not Applicable — reproduced against the unmodified official images.

Anything else?

This is separate from #52368 / #65136 / #65161 (uncaught task exceptions not captured by
Sentry), which is fixed in 3.3.0. This before_send/transport resolution bug is a
different issue, still present unchanged in 3.3.0rc2.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct