Fix execution API server URL handling for relative paths (#49782)

Follow-up of #49747 which broke some workflows including `breeze start-airflow`, requiring #49753.

When base_url is a relative path (starts with '/'), default to using
http://localhost:8080 as the base URL. This maintains backward compatibility
with the previous default behavior while still allowing custom absolute URLs
through either `base_url` or `execution_api_server_url` configuration.

Fixes `httpx.UnsupportedProtocol` error (as seen in #49753) that occurred when base_url was
a relative path without protocol.

(cherry picked from commit 3b9ae48)

Author: kaxil

airflow-core/src/airflow/config_templates/config.yml

@@ -537,7 +537,8 @@ core:
     execution_api_server_url:
       description: |
         The url of the execution api server. Default is ``{BASE_URL}/execution/``
-        where ``{BASE_URL}`` is the base url of the API Server.
+        where ``{BASE_URL}`` is the base url of the API Server. If ``{BASE_URL}`` is not set,
+        it will use ``http://localhost:8080`` as the default base url.
       version_added: 3.0.0
       type: string
       example: ~

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

@@ -110,6 +110,9 @@ def _execute_work(log: logging.Logger, workload: workloads.ExecuteTask) -> None:
     setproctitle(f"airflow worker -- LocalExecutor: {workload.ti.id}")
 
     base_url = conf.get("api", "base_url", fallback="/")
+    # If it's a relative URL, use localhost:8080 as the default
+    if base_url.startswith("/"):
+        base_url = f"http://localhost:8080{base_url}"
     default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
 
     # This will return the exit code of the task process, but we don't care about that, just if the

airflow-core/tests/unit/executors/test_local_executor.py

@@ -185,8 +185,17 @@ def test_clean_stop_on_signal(self):
                 },
                 "http://custom-server/execution/",
             ),
+            ({}, "http://localhost:8080/execution/"),
+            ({("api", "base_url"): "/"}, "http://localhost:8080/execution/"),
+            ({("api", "base_url"): "/airflow/"}, "http://localhost:8080/airflow/execution/"),
+        ],
+        ids=[
+            "base_url_fallback",
+            "custom_server",
+            "no_base_url_no_custom",
+            "base_url_no_custom",
+            "relative_base_url",
         ],
-        ids=["base_url_fallback", "custom_server"],
     )
     @mock.patch("airflow.sdk.execution_time.supervisor.supervise")
     def test_execution_api_server_url_config(self, mock_supervise, conf_values, expected_server):

providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py

@@ -159,6 +159,9 @@ def execute_workload(input: str) -> None:
     log.info("[%s] Executing workload in Celery: %s", celery_task_id, workload)
 
     base_url = conf.get("api", "base_url", fallback="/")
+    # If it's a relative URL, use localhost:8080 as the default
+    if base_url.startswith("/"):
+        base_url = f"http://localhost:8080{base_url}"
     default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
 
     supervise(

providers/edge3/src/airflow/providers/edge3/cli/edge_command.py

@@ -272,6 +272,9 @@ def _run_job_via_supervisor(
 
             try:
                 base_url = conf.get("api", "base_url", fallback="/")
+                # If it's a relative URL, use localhost:8080 as the default
+                if base_url.startswith("/"):
+                    base_url = f"http://localhost:8080{base_url}"
                 default_execution_api_server = f"{base_url.rstrip('/')}/execution/"
 
                 supervise(