[HyukjinKwon]

What changes were proposed in this pull request?

Bandaid for the arrow-UDF parity hang (SPARK-59032). Move the Rust behavioral
e2e tests out of the parity job into their own connect-e2e job with its own Connect
server, so the parity gate starts against a server nothing else has touched.

Root cause (why this helps): the parity gate hung on
test_parity_arrow_python_udf.py only when the behavioral e2e cargo tests had already
run against the same shared Connect server. test_python_udf_segfault deliberately
crashes a Python worker (ctypes.string_at(0), faulthandler enabled); on a server the
e2e tests had already stressed, that left the server's Python-worker reuse wedged, so a
later UDF query (test_raise_stop_iteration's range(5)) never produced results - the
server streamed only ExecutionProgress responses (0 Arrow batches, no ResultComplete),
and the reattachable client dutifully re-attached forever. Diagnosed from a per-test
client thread dump + server jstack + a response-type histogram of the stalled streams.

Giving the parity run a clean server sidesteps the wedge. No test is skipped. This is
an infra-level mitigation; a proper client/server worker-recovery fix is a follow-up.

Also piggybacked here (unrelated cleanup, from #63):

• Remove compiler-flagged unused Rust imports and redundant StandardScalerModel
  fields, plus unused imports/vars in tests (a clippy --all-targets sweep). No
  behavior change.

Why are the changes needed?

The parity gate timed out on every run because of the shared-server interaction above.
This makes it green without skipping any test.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

CI: the parity gate runs against a fresh server; the behavioral e2e tests run in their
own connect-e2e job. cargo build --release is clean and the apache-spark-connect
unit tests pass.

This pull request and its description were written by Isaac.

[viirya]

Verified the core mechanism against the tonic 0.14.6 source rather than taking the description's word for it, and it holds up: http2_keep_alive_interval / keep_alive_timeout / keep_alive_while_idle do reach hyper (transport/channel/service/connection.rs:38-51), an expired client PING surfaces as hyper::Error::is_timeout(), and Status::from_hyper_error maps exactly that to UNAVAILABLE (status.rs:437-440), which can_retry accepts (retries.rs:74) — as does the reference DefaultPolicy, which matters here since the reattach loop driving the parity gate is the reference client's, not reattach.rs. Fixing a documented-but-ignored option is clearly right on its own merits. A few things before this merges.

keepalive_timeout_ms is silently coerced rather than validated. int_param returns i64 and accepts any parsable integer, so the two fields behave inconsistently: grpc_keepalive_time_ms=-1 quietly disables keepalive via the guard, but grpc_keepalive_timeout_ms=-1 becomes a zero-duration timeout through .max(0). Every PING then times out immediately, so the first keepalive interval kills a healthy connection with UNAVAILABLE, and under the 15-retry default that's a retry storm rather than a clean error — strictly worse than the hang being fixed. .max(0) is what disguises it, turning a bad input into a plausible-looking value instead of rejecting it. A zero timeout is never a meaningful request; either reject non-positive values in ChannelBuilder (consistent with how session_id and user_agent already validate) or clamp to a sane floor. Worth a unit test for the dangerous combination specifically — a valid interval with a -1 timeout — since keepalive_override only covers the happy path.

The same bug-class survives one line above the fix. The defect here is "a documented channel option is parsed but never applied," and channel.rs:21 declares GRPC_MAX_MESSAGE_LENGTH_DEFAULT = 128 MiB which is referenced nowhere in the crate — no max_decoding_message_size / max_encoding_message_size call exists anywhere. tonic defaults to a 4 MiB receive cap (codec/mod.rs:101), so any collect() returning a >4 MiB Arrow batch fails with a decode error the reference client handles fine. It's a dead constant documenting the intent, i.e. exactly the shape of the bug just fixed. Two builder calls on the same ep would close it; otherwise worth a follow-up ticket so it isn't rediscovered later as a mystery decode error.

The description covers two of the three changes. Removing the diagnostic workflow is accurate for .github/workflows, but 31ac6ee and 509b9e6 add a new permanent diagnostic layer the description never mentions — the faulthandler watchdog in rust_transport_plugin.py and _dump_server_jvm() in run_official_tests.py. Please fold that into the description; as it stands an approval based on the description wouldn't know it's in the diff. The machinery itself is reasonable to keep (a CI-only hang with no local repro genuinely needs post-mortem capture), with three things I'd tighten:

• _dump_server_jvm takes found[0] blindly, but pgrep -f org.apache.spark.deploy.SparkSubmit can match several PIDs, so it may jstack the wrong process. found[-1], or dumping every match, is more robust for a diagnostic whose whole value is landing on the right JVM.
• run_ours is called from a ThreadPoolExecutor, so with --jobs > 1 a timeout in file A dumps a JVM shared with B and C while attributing it to A alone. The per-file dump_path is correctly keyed by test_file.name; the JVM dump has no such keying, so a comment noting it's only unambiguous at --jobs 1 would help.
• _hang_dump_fh is a module-scope open(..., "w") that's never closed, so an unwritable RUST_HANG_DUMP raises during plugin import — failing the whole file with rc≠0 and looking like a client regression. A try/except there applies the same principle you already got right in _dump_server_jvm ("must never mask the timeout").

Two different root causes are now in play. Replacing the whole-file skip with a retry is the right call, and identifying the SLOW_FILE_TIMEOUTS=1200 attempt as wrong and reverting it is the honest read. But if keepalive genuinely fixes the root cause, the 60s/20s defaults surface a wedged stream in ~80s, well inside the 360s cap, so the retry should never fire. Meanwhile SPARK-59032 still says "suspected race opening the first reattachable-execute stream" while the new client comment asserts the keepalive explanation. Keepalive explains a wedged transport; it doesn't explain 509b9e6's own observation that the server produced nothing at all while the reference client got an error in <1s. Worth stating explicitly that keepalive makes the symptom recoverable without necessarily being the root cause, so the ticket doesn't get closed on a fix that only masks it. Separately, note that adding a hang to FLAKY_FILES costs a full 360s cap per attempt — a recurrence burns 720s+ of gate time. I agree with the trade (coverage over speed), just flagging that this file will make the gate slow before it makes it red.

Smaller things: the lazy-connect rationale now appears twice in connect(), at client.rs:82-87 and again at client.rs:121-125, saying the same thing in different words — the refactor moved the call without removing the original comment, so one should go. The True → "env" change in the work() tuple is churn: the value is only consumed truthily (if skipped_file:) and the field is named skipped_file, so a magic string implies a variant set that doesn't exist while the new inline comment already does the whole job. format!("Invalid endpoint: {}", e) keeps the positional form while the new code above it uses inline {scheme} captures. And the keepalive comment is 10 lines of incident narrative over 8 lines of code — the why (parsed but never applied; expired PING → UNAVAILABLE → retriable) is worth keeping, but the CI-incident detail is already in the commit message and the JIRA.

[HyukjinKwon]

Thanks @viirya — thorough review, all fair. Addressed in the latest commit (866acf0):

• keepalive timeout coercion. Fixed the real bug: .max(0) is gone. The logic now lives in a validated ChannelBuilder::keepalive() that returns None (never a zero-duration) when the interval or timeout is non-positive — so grpc_keepalive_timeout_ms=-1 disables keepalive, consistent with how a -1 interval already behaves, instead of producing an immediately-expiring PING. Added a unit test for exactly the (valid-interval, -1-timeout) combination.
• max message length (128 MiB parsed but never applied). Agreed it's the same bug-class and a real >4 MiB decode hazard. Filed as a follow-up: SPARK-59037, so it isn't rediscovered as a mystery decode error.
• diagnostic layer in the description. Folded the faulthandler watchdog + _dump_server_jvm into the PR description explicitly. Also tightened all three: found[-1] instead of [0], raised the jstack cap 20k→60k (the first run truncated exactly the execute/PythonRunner threads), a comment that the JVM dump is only unambiguous at --jobs 1, and a try/except around the dump-file open so an unwritable RUST_HANG_DUMP can't fail plugin import.
• two root causes / framing. Fully agree. Reframed both the PR and SPARK-59032: keepalive makes the symptom recoverable but does not explain "server produced nothing while the reference got an error in <1s," so the ticket stays open for the true cause (the added jstack targets exactly that). Noted the FLAKY_FILES retry is a 360s-cap-per-attempt trade — coverage over speed.
• smaller things. Removed the duplicated lazy-connect comment, reverted the skipped_file "env" string back to a bool, switched to inline {e} captures, and trimmed the keepalive comment to the why.

[HyukjinKwon]

Update: the actual root cause is now found and fixed, and the PR has pivoted accordingly (title/description updated).

It was not the keepalive path and not test-specific. Via CI RPC-tracing + a server jstack: the reattachable-execute driver fires a ReleaseExecute after every response on a background pool, and each release closure captures the operation's iterator. Our release_execute was unbounded, so a stalled release kept its future pending → kept the iterator alive → kept the iterator's open ExecutePlan HTTP/2 stream alive; enough of those accumulate on the single shared connection that a later ExecutePlan can't get a stream and the client blocks forever while the server is fully idle. Cumulative → it landed on different tests depending on timing (test_raise_stop_iteration, test_udf_in_generate, …).

Fix: bound release_execute with a 30s deadline. ReleaseExecute is best-effort — reattach.py runs it on a pool and logs-and-ignores failures ("server is equipped to deal with abandoned executions", identical on v4.2.0 and master) — so a timed-out release frees its thread and closes its stream, nothing accumulates. Verified in CI: with the hang test not deselected, test_parity_arrow_python_udf.py passes 319/320 and all arrow files pass. The keepalive change (which you reviewed) stays as an independent robustness fix; all the earlier workarounds are removed.