user-facing phrasing for Niko

Author: ferruzzi

airflow-core/src/airflow/executors/base_executor.py

@@ -241,7 +241,10 @@ def queue_workload(self, workload: workloads.All, session: Session) -> None:
                 )
             self.queued_callbacks[workload.callback.id] = workload
         else:
-            raise ValueError(f"Un-handled workload kind {type(workload).__name__!r} in {type(self).__name__}")
+            raise ValueError(
+                f"Un-handled workload type {type(workload).__name__!r} in {type(self).__name__}. "
+                f"Workload must be one of: ExecuteTask, ExecuteCallback."
+            )
 
     def _get_workloads_to_schedule(self, open_slots: int) -> list[tuple[WorkloadKey, workloads.All]]:
         """

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -379,14 +379,14 @@ def _get_workload_team_name(self, workload: SchedulerWorkload, session: Session)
 
         if team_name:
             self.log.debug(
-                "Resolved team name '%s' for workload %s (dag_id=%s)",
+                "Resolved team name '%s' for task or callback %s (dag_id=%s)",
                 team_name,
                 workload,
                 dag_id,
             )
         else:
             self.log.debug(
-                "No team found for workload %s (dag_id=%s) - DAG may not have bundle or team association",
+                "No team found for task or callback %s (dag_id=%s) - DAG may not have bundle or team association",
                 workload,
                 dag_id,
             )
@@ -3268,15 +3268,17 @@ def _try_to_load_executor(
                         executor = _executor
 
         if executor is not None:
-            self.log.debug("Found executor %s for workload %s (team: %s)", executor.name, workload, team_name)
+            self.log.debug(
+                "Found executor %s for task or callback %s (team: %s)", executor.name, workload, team_name
+            )
         else:
             # This case should not happen unless some (as of now unknown) edge case occurs or direct DB
             # modification, since the DAG parser will validate the tasks in the DAG and ensure the executor
             # they request is available and if not, disallow the DAG to be scheduled.
             # Keeping this exception handling because this is a critical issue if we do somehow find
             # ourselves here and the user should get some feedback about that.
             self.log.warning(
-                "Executor, %s, was not found but a workload was configured to use it",
+                "Executor, %s, was not found but a Task or Callback was configured to use it",
                 workload.get_executor_name(),
             )
 

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -8220,7 +8220,8 @@ def test_multi_team_try_to_load_executor_explicit_executor_team_mismatch(
 
             # Should log a warning when no executor is found
             mock_log.warning.assert_called_once_with(
-                "Executor, %s, was not found but a workload was configured to use it", "secondary_exec"
+                "Executor, %s, was not found but a Task or Callback was configured to use it",
+                "secondary_exec"
             )
 
         # Should return None since we failed to resolve an executor due to the mismatch. In practice, this