AsyncSparkCursor: handle executor cleanup when session termination fails #795

Parent: #791

Problem

AsyncSparkCursor.close() calls super().close() before _executor.shutdown().
If session termination raises OperationalError, shutdown is never reached.
A cursor that has submitted work can retain its executor and polling tasks after the failed close.

Evidence: close implementation and termination errors.

Reproduction

A deterministic probe can isolate the branch without AWS:

from unittest.mock import MagicMock
from pyathena.error import OperationalError
from pyathena.spark.async_cursor import AsyncSparkCursor

cursor = AsyncSparkCursor.__new__(AsyncSparkCursor)
cursor._terminate_session = MagicMock(side_effect=OperationalError("termination failed"))
cursor._executor = MagicMock()
try:
    cursor.close()
except OperationalError:
    pass
assert cursor._executor.shutdown.call_count == 0

This reproduced the skipped shutdown branch; it does not measure a live thread or remote-session leak.

Environment

Checked at a03bc2b4ea5dddcbf6230aafda0a359dce12ef0e with Python 3.13.1.
Evidence below is from source inspection and offline probes with AWS calls replaced by mocks; no live Athena validation was performed.

Proposed fix (optional)

Define cleanup ordering and failure behavior so a remote termination error cannot silently skip local executor cleanup.
Preserve close(wait=...) behavior and report the remote error.
Define what happens to queued/running work, repeated close calls, and a later retry of unfinished remote cleanup.
Do not mask termination failures or claim a running worker has stopped merely because shutdown(wait=False) returned.

Acceptance:

• Cover termination success/failure, queued/running futures, wait=True/False, and repeated close.
• Use deterministic synchronization to check worker cleanup and exception propagation.
• Verify that no unrelated calculation or externally owned session is stopped by the cleanup policy.