Fix scheduler crash with email notifications (apache#56429)

The ``EmailNotificationRequest`` class name (25 characters) exceeded the
database constraint for ``DbCallbackRequest.callback_type column`` (20
characters), causing scheduler crashes when email notifications were
triggered for task failures or retries.

This fix renames the class to ``EmailRequest`` (12 characters) to fit within
the constraint. A backwards compatibility alias ensures existing DB
entries with `'EmailNotificationRequest'` can still be deserialized via
getattr lookup.

The 20-character limit is arbitrary and does not affect performance.
In a follow-up PR for 3.2, we should increase this to 50+ characters
to accommodate descriptive class names without requiring abbreviations.

Fixes apache#56426

Author: kaxil

airflow-core/src/airflow/callbacks/callback_requests.py

@@ -77,7 +77,7 @@ def is_failure_callback(self) -> bool:
         }
 
 
-class EmailNotificationRequest(BaseCallbackRequest):
+class EmailRequest(BaseCallbackRequest):
     """Email notification request for task failures/retries."""
 
     ti: ti_datamodel.TaskInstance
@@ -86,7 +86,7 @@ class EmailNotificationRequest(BaseCallbackRequest):
     """Whether this is for a failure or retry email"""
     context_from_server: ti_datamodel.TIRunContext
     """Task execution context from the Server"""
-    type: Literal["EmailNotificationRequest"] = "EmailNotificationRequest"
+    type: Literal["EmailRequest"] = "EmailRequest"
 
 
 class DagRunContext(BaseModel):
@@ -108,6 +108,9 @@ class DagCallbackRequest(BaseCallbackRequest):
 
 
 CallbackRequest = Annotated[
-    DagCallbackRequest | TaskCallbackRequest | EmailNotificationRequest,
+    DagCallbackRequest | TaskCallbackRequest | EmailRequest,
     Field(discriminator="type"),
 ]
+
+# Backwards compatibility alias
+EmailNotificationRequest = EmailRequest

airflow-core/src/airflow/dag_processing/processor.py

@@ -31,7 +31,7 @@
 from airflow.callbacks.callback_requests import (
     CallbackRequest,
     DagCallbackRequest,
-    EmailNotificationRequest,
+    EmailRequest,
     TaskCallbackRequest,
 )
 from airflow.configuration import conf
@@ -243,7 +243,7 @@ def _execute_callbacks(
             _execute_task_callbacks(dagbag, request, log)
         elif isinstance(request, DagCallbackRequest):
             _execute_dag_callbacks(dagbag, request, log)
-        elif isinstance(request, EmailNotificationRequest):
+        elif isinstance(request, EmailRequest):
             _execute_email_callbacks(dagbag, request, log)
 
 
@@ -354,9 +354,7 @@ def get_callback_representation(callback):
             log.exception("Error in callback at index %d: %s", idx, callback_repr)
 
 
-def _execute_email_callbacks(
-    dagbag: DagBag, request: EmailNotificationRequest, log: FilteringBoundLogger
-) -> None:
+def _execute_email_callbacks(dagbag: DagBag, request: EmailRequest, log: FilteringBoundLogger) -> None:
     """Execute email notification for task failure/retry."""
     dag = dagbag.dags[request.ti.dag_id]
     task = dag.get_task(request.ti.task_id)

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

@@ -43,7 +43,7 @@
 from airflow.callbacks.callback_requests import (
     DagCallbackRequest,
     DagRunContext,
-    EmailNotificationRequest,
+    EmailRequest,
     TaskCallbackRequest,
 )
 from airflow.configuration import conf
@@ -959,7 +959,7 @@ def process_executor_events(
                         "Sending email request for task %s to DAG Processor",
                         ti,
                     )
-                    email_request = EmailNotificationRequest(
+                    email_request = EmailRequest(
                         filepath=ti.dag_model.relative_fileloc,
                         bundle_name=ti.dag_version.bundle_name,
                         bundle_version=ti.dag_version.bundle_version,

airflow-core/tests/unit/callbacks/test_callback_requests.py

@@ -32,7 +32,7 @@
     CallbackRequest,
     DagCallbackRequest,
     DagRunContext,
-    EmailNotificationRequest,
+    EmailRequest,
     TaskCallbackRequest,
 )
 from airflow.models.dag import DAG
@@ -314,9 +314,9 @@ def test_dag_callback_request_serialization_with_context(self):
         assert result.context_from_server.last_ti.task_id == "test_task"
 
 
-class TestEmailNotificationRequest:
+class TestEmailRequest:
     def test_email_notification_request_serialization(self):
-        """Test EmailNotificationRequest can be serialized and used in CallbackRequest union."""
+        """Test EmailRequest can be serialized and used in CallbackRequest union."""
         ti_data = TIDataModel(
             id=str(uuid.uuid4()),
             task_id="test_task",
@@ -331,8 +331,8 @@ def test_email_notification_request_serialization(self):
 
         current_time = timezone.utcnow()
 
-        # Create EmailNotificationRequest
-        email_request = EmailNotificationRequest(
+        # Create EmailRequest
+        email_request = EmailRequest(
             filepath="/path/to/dag.py",
             bundle_name="test_bundle",
             bundle_version="1.0.0",
@@ -359,17 +359,17 @@ def test_email_notification_request_serialization(self):
 
         # Test serialization
         json_str = email_request.to_json()
-        assert "EmailNotificationRequest" in json_str
+        assert "EmailRequest" in json_str
         assert "failure" in json_str
 
         # Test deserialization
-        result = EmailNotificationRequest.from_json(json_str)
+        result = EmailRequest.from_json(json_str)
         assert result == email_request
         assert result.email_type == "failure"
         assert result.ti.task_id == "test_task"
 
     def test_callback_request_union_with_email_notification(self):
-        """Test EmailNotificationRequest works in CallbackRequest union type."""
+        """Test EmailRequest works in CallbackRequest union type."""
         ti_data = TIDataModel(
             id=str(uuid.uuid4()),
             task_id="test_task",
@@ -402,7 +402,7 @@ def test_callback_request_union_with_email_notification(self):
         )
 
         email_data = {
-            "type": "EmailNotificationRequest",
+            "type": "EmailRequest",
             "filepath": "/path/to/dag.py",
             "bundle_name": "test_bundle",
             "bundle_version": "1.0.0",
@@ -416,7 +416,7 @@ def test_callback_request_union_with_email_notification(self):
         adapter = TypeAdapter(CallbackRequest)
         callback_request = adapter.validate_python(email_data)
 
-        # Verify it's correctly identified as EmailNotificationRequest
-        assert isinstance(callback_request, EmailNotificationRequest)
+        # Verify it's correctly identified as EmailRequest
+        assert isinstance(callback_request, EmailRequest)
         assert callback_request.email_type == "retry"
         assert callback_request.ti.task_id == "test_task"

airflow-core/tests/unit/dag_processing/test_processor.py

@@ -45,7 +45,7 @@
     CallbackRequest,
     DagCallbackRequest,
     DagRunContext,
-    EmailNotificationRequest,
+    EmailRequest,
     TaskCallbackRequest,
 )
 from airflow.dag_processing.dagbag import DagBag
@@ -1303,7 +1303,7 @@ def test_execute_email_callbacks_failure(self, mock_send_email):
         )
 
         current_time = timezone.utcnow()
-        request = EmailNotificationRequest(
+        request = EmailRequest(
             filepath="/path/to/dag.py",
             bundle_name="test_bundle",
             bundle_version="1.0.0",
@@ -1363,7 +1363,7 @@ def test_execute_email_callbacks_retry(self, mock_send_email):
 
         current_time = timezone.utcnow()
 
-        request = EmailNotificationRequest(
+        request = EmailRequest(
             filepath="/path/to/dag.py",
             bundle_name="test_bundle",
             bundle_version="1.0.0",
@@ -1422,7 +1422,7 @@ def test_execute_email_callbacks_no_email_configured(self, mock_send_email):
         )
 
         current_time = timezone.utcnow()
-        request = EmailNotificationRequest(
+        request = EmailRequest(
             filepath="/path/to/dag.py",
             bundle_name="test_bundle",
             bundle_version="1.0.0",
@@ -1480,7 +1480,7 @@ def test_execute_email_callbacks_email_disabled_for_type(self, mock_send_email):
         current_time = timezone.utcnow()
 
         # Create request for failure (but email_on_failure is False)
-        request = EmailNotificationRequest(
+        request = EmailRequest(
             filepath="/path/to/dag.py",
             bundle_name="test_bundle",
             bundle_version="1.0.0",