[Enhancement] Allow Spark job and session log streaming to wait for the driver pod

[vivek-gite]

What you would like to be added?

Description

Spark job and Spark Connect session log streaming can fail when requested immediately after creating the resource.

Both get_job_logs() and get_session_logs() retrieve the resource once and raise a RuntimeError when driver_pod_name is not yet present. However, Kubernetes normally creates the SparkApplication or SparkConnect resource before the Spark Operator creates the driver pod and updates the resource status.

As a result, this natural workflow is unreliable:

job = client.submit_job(...)

for line in client.get_job_logs(job.name, follow=True):
    print(line)

Depending on timing, the call may fail with a “No driver pod” error even though the Spark workload is still starting normally.

Current behavior

1. The Spark resource is created.
2. The user immediately requests its logs.
3. The resource status does not contain a driver pod name yet.
4. get_job_logs() or get_session_logs() raises RuntimeError.
5. The user must implement custom polling and retry logic before using the SDK’s log API.

Expected behavior

The SDK should optionally wait for the driver pod to become available before retrieving or following its logs.

Existing behavior should remain available to preserve backward compatibility and support callers that prefer an immediate response.

Proposed API

Add keyword-only waiting parameters to both log methods:

client.get_job_logs(
    name,
    follow=False,
    *,
    wait_for_driver=False,
    timeout=120,
    polling_interval=2,
)

client.get_session_logs(
    name,
    follow=False,
    *,
    wait_for_driver=False,
    timeout=120,
    polling_interval=2,
)

Using keyword-only parameters preserves the existing positional API. Keeping wait_for_driver=False as the default also preserves current behavior.

When wait_for_driver=True, the backend should poll the resource until one of the following occurs:

• The driver pod name becomes available, after which log retrieval begins.
• The resource reaches a terminal failure before a driver pod is created.
• The resource is deleted.
• The configured timeout expires.

Error handling

Errors should include enough context to help users diagnose the failure:

• Resource kind, namespace, and name.
• Last observed resource state.
• Configured timeout when applicable.
• Whether the resource failed, was deleted, or timed out while waiting.

Example:

Timed out waiting for the driver pod for SparkApplication default/example-job
after 120 seconds (last observed state: Submitted).

Implementation notes

• Reuse a shared internal helper for Spark jobs and Spark Connect sessions to avoid duplicating polling logic.
• Use time.monotonic() for timeout calculations.
• Validate that timeout and polling_interval are positive.
• Begin reading logs only after a non-empty driver pod name is available.
• Do not make additional Kubernetes requests when wait_for_driver=False.
• Preserve the existing lazy Iterator[str] return contract.

Acceptance criteria

• Existing calls without the new parameters retain their current behavior.
• wait_for_driver=True waits for delayed driver pod creation and then retrieves logs.
• Both get_job_logs() and get_session_logs() support the behavior.
• A terminal failure before pod creation produces an actionable error.
• Resource deletion while waiting produces an actionable error.
• Timeout errors include the resource identity, timeout, and last observed state.
• Invalid timeout or polling values are rejected with a clear ValueError.
• Unit tests mock all Kubernetes interactions and make no network calls.
• Tests cover immediate pod availability, delayed availability, failure, deletion, and timeout.
• Public docstrings and a user-facing example are updated.

Why is this needed?

Waiting for logs is a common part of submitting and debugging Spark workloads. Handling the driver-pod creation delay inside SparkClient removes timing-dependent failures and prevents every user from having to implement the same polling logic.

Love this feature?

Give it a 👍 We prioritize the features with most 👍

[vivek-gite]

/assign