Fix DagRun._emit_dagrun_span crash on None/empty context_carrier

[yulit0738]

Summary

Fixes a crash in DagRun._emit_dagrun_span() when context_carrier is None or empty {}.

• Symptom: AttributeError: 'NoneType' object has no attribute 'get' in TraceContextTextMapPropagator().extract()
• Cause: DagRuns created before OTel tracing was enabled have context_carrier = NULL in the database. When these DagRuns complete, _emit_dagrun_span() passes None directly to extract().
• Fix: Early return when context_carrier is falsy, consistent with the existing guard in task_runner.py:148.

Reproduction Path

1. Have existing DagRuns in the database (created before OTel was enabled → context_carrier = NULL)
2. Enable OTel tracing
3. Those DagRuns complete → update_state() → _emit_dagrun_span() → crash

Note: This supersedes #61655 which targeted OtelTrace.extract() — that class was since removed in #63452.

Testing

• Added parametrized unit test covering both None and {} carrier values
• Verified no span is emitted when carrier is missing (graceful skip)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Generated-by: Claude (Anthropic) following the guidelines

[uranusjr]

TraceContextTextMapPropagator().extract() already has logic to deal with None and empty dict. I am under the impression these are expected values that should go through the same code path, not silently excluded early.

[yulit0738]

TraceContextTextMapPropagator().extract() already has logic to deal with None and empty dict. I am under the impression these are expected values that should go through the same code path, not silently excluded early.

@uranusjr verified against the latest opentelemetry-python main — TraceContextTextMapPropagator.extract() has no None guard for carrier. DefaultGetter.get() calls
carrier.get(key, None) directly, which raises AttributeError on None.

The context_carrier column is nullable=True. After upgrading Airflow to a version with OTel tracing, existing DagRuns in the DB have context_carrier = NULL. When those DagRuns complete,
the scheduler crashes in _emit_dagrun_span() — not a task failure, but a scheduler process crash.

task_runner.py:148 already uses the same pattern: if msg.ti.context_carrier else None.

I can narrow the guard to only None (not empty dict) if you prefer — {} does pass through extract() safely.

[uranusjr]

Let’s pass self.context_carrier or {} to OTel then, that’s probably more idiomatic.

[yulit0738]

@uranusjr
Updated — now using self.context_carrier or {} instead of the early return, so OTel receives a valid carrier and emits a root span per spec.

You were right that this should go through the same code path. The only issue was None specifically — DefaultGetter.get() calls carrier.get(key, None) which raises AttributeError on None. The or {} lets OTel handle it naturally.