refactor(yarn_client): use Location header verbatim, drop urljoin

YARN's 307 Location header is always a fully-qualified absolute URL.
The previous implementation used urllib.parse.urljoin to handle the
hypothetical case of a relative Location, but that's a defensive
codepath for a scenario that does not occur in practice. Drop it:

  - Remove 'from urllib.parse import urljoin'.
  - _request_following_redirects now assigns the Location value
    straight to the next request URL.
  - Update the docstring to reflect the new contract.
  - Remove the test that exercised the relative-Location path.

Full suite 246 passed (1 fewer than the previous commit, matching the
test that was removed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

spark_executor/core/yarn_client.py

@@ -28,7 +28,6 @@ field was designed for, but no longer requires the `yarn` CLI to interpret it.
 """
 import json
 from dataclasses import dataclass
-from urllib.parse import urljoin
 
 import httpx
 import httpx_kerberos
@@ -159,9 +158,9 @@ def _request_following_redirects(
     request lands unauthenticated and gets 401/403. Doing it ourselves
     keeps the auth + verify config on every hop.
 
-    Relative `Location` values are resolved against the current request
-    URL. A 3xx without a `Location` header is returned as-is (the caller
-    will treat it as a non-2xx response and surface a clean error).
+    YARN always returns a fully-qualified absolute URL in `Location`, so
+    we use it verbatim. A 3xx without a `Location` header is treated as
+    a misconfigured server and raises YarnError.
     """
     for hop in range(max_redirects + 1):
         resp = _request(method, url, **kwargs)
@@ -172,9 +171,8 @@ def _request_following_redirects(
             raise YarnError(
                 f"YARN {method} returned {resp.status_code} with no Location header at {url}"
             )
-        next_url = urljoin(url, location)
-        logger.debug(f"YARN {method} {url} -> {resp.status_code}, following to {next_url}")
-        url = next_url
+        logger.debug(f"YARN {method} {url} -> {resp.status_code}, following to {location}")
+        url = location
     raise YarnError(
         f"YARN {method} exceeded {max_redirects} redirects (last url: {url})"
     )

tests/unit/test_yarn_client.py

@@ -274,30 +274,6 @@ def test_logs_follows_redirect_chain_then_200():
     ]
 
 
-def test_logs_relative_location_resolved_against_request_url():
-    """A relative Location header (e.g. '/redirected/stdout') must be
-    resolved against the request URL, not treated as an absolute path."""
-    responses = [
-        _resp(404),  # aggregated-logs missing
-        _resp(
-            200,
-            json_data={
-                "app": {
-                    "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs",
-                }
-            },
-        ),
-        _resp(307, headers={"Location": "/redirected/stdout"}),
-        _resp(200, text="resolved body"),
-    ]
-    with patch(
-        "spark_executor.core.yarn_client.httpx.request", side_effect=responses
-    ) as m:
-        out = get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM))
-    assert out == "resolved body"
-    assert m.call_args_list[3].args == ("GET", "http://nm1:8042/redirected/stdout")
-
-
 def test_logs_5xx_on_aggregated_endpoint_raises_immediately():
     with patch(
         "spark_executor.core.yarn_client.httpx.request",