Conditionally add OTEL events when processing executor events (#43558) (

#43567)

It's possible that the start/end date are null when processing an
executor event, and there is no point in adding an OTEL event in that
case.

Before this, we'd try and convert `None` to nanoseconds and blow up the
scheduler.

Note: I don't think `queued_dttm` can be empty, but figured it didn't hurt to
guard against it just in case I've overlooked a way it can be possible.

(cherry picked from commit fe41e15)

airflow/jobs/scheduler_job_runner.py

@@ -847,9 +847,12 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->
                 span.set_attribute("ququed_by_job_id", ti.queued_by_job_id)
                 span.set_attribute("pid", ti.pid)
                 if span.is_recording():
-                    span.add_event(name="queued", timestamp=datetime_to_nano(ti.queued_dttm))
-                    span.add_event(name="started", timestamp=datetime_to_nano(ti.start_date))
-                    span.add_event(name="ended", timestamp=datetime_to_nano(ti.end_date))
+                    if ti.queued_dttm:
+                        span.add_event(name="queued", timestamp=datetime_to_nano(ti.queued_dttm))
+                    if ti.start_date:
+                        span.add_event(name="started", timestamp=datetime_to_nano(ti.start_date))
+                    if ti.end_date:
+                        span.add_event(name="ended", timestamp=datetime_to_nano(ti.end_date))
                 if conf.has_option("traces", "otel_task_log_event") and conf.getboolean(
                     "traces", "otel_task_log_event"
                 ):