fix: scope Celeborn bootstrap hooks to Comet clients

[pingzh]

Which issue does this PR close?

Closes #5529.

Follow-up to #5352.

Rationale for this change

Comet installs a bootstrap hook in Celeborn's shared client factory to track transport ownership. Failures while installing or invoking that hook could escape into Celeborn client creation and disrupt ordinary, non-Comet Celeborn traffic. The hook also remained installed after the final Comet client released it.

What changes are included in this PR?

• Treat reflective, I/O, and runtime hook instrumentation failures as non-fatal, disable precise transport tracking, and fall back to Celeborn PushState accounting.
• Reference-count the shared hook and remove the exact proxy when the final Comet client releases it, while preserving unrelated Celeborn bootstraps and their order.
• Synchronize hook release with active bootstrap calls, keep stale bootstrap snapshots inert, and restore tracking when a later Comet client reacquires the shared factory.
• Preserve already-observed transport ownership during fallback and serialize raw-push state binding so executor admission is never released early.

How are these changes tested?

• Added regression coverage for bootstrap instrumentation failures, ordinary Celeborn client creation, shared hook ownership, final-owner removal, stale snapshots, reacquisition, fatal bootstrap errors, and concurrent hook release.
• Added fallback accounting coverage for recreated push states, concurrent submissions, cleanup races, retries, callbacks, and pending Netty writes.
• Ran CelebornShufflePartitionPusherSuite with the default Spark 4.1 profile: 53 tests passed.
• Ran Spotless and the Scala 2.12 parser check successfully.

[ajsquared]

Reviewed 36f418e. One non-blocking P2 accounting issue below; no P1 findings. Static review only.

[ajsquared]

Reviewed 36f418e. The previously posted P2 feedback is non-blocking.

[ziting-openai]

Reviewed 36f418e61672243ab90651f462ea7b34f7e44009 using spark-pr-review-memo. No P1 findings. One existing significant P2 remains: fallback admission after cancellation and failed transport. I added isolated reproduction evidence to that thread. This approval follows the requested zero-P1/fewer-than-three-P2 threshold.

Validation included unchanged current-head/parent Java probes with a Celeborn behavior fixture and logging shim, plus public JVM descriptor comparison and pinned Celeborn source review. I did not run the full Comet suite. Exact-head CI has one macOS scan JVM SIGSEGV whose connection to this patch is unestablished.

[ 🤖 posted by Codex on behalf of ziting-openai using the spark-pr-review-memo skill 🤖 ]

[ziting-openai]

Re-reviewed 9a75f9de3e37136968694a74a5d6ad2623780385 using spark-pr-review-memo. No P1 findings. The single-request fallback cancellation and cancellation-before-downgrade cases are fixed. One existing significant P2 remains when several already submitted precise requests are downgraded together: current-head reproduction. This approval follows the requested zero-P1/fewer-than-three-P2 threshold.

Validation: isolated unchanged-source Java probes compared this head with the prior reviewed head, including positive controls and the remaining multiple-request leak. Public partition-pusher JVM descriptors are unchanged. The probe used transport fixtures and a logging shim on JDK 21; I did not run the full Comet/Celeborn/Netty suite. Current-head CI is still in progress with no reported failures at preflight, and the released Celeborn 0.6.0/0.7.0 reflection checks passed.

[ 🤖 posted by Codex on behalf of ziting-openai using the spark-pr-review-memo skill 🤖 ]

[sunchao]

cc @andygrove

[sunchao]

Reviewed 9a75f9de3e37136968694a74a5d6ad2623780385 against 1e10eedd6e0303adcac4573f44d5c07ffb0fbacd. No verified P1/P2 findings. This was a source review, including the existing fallback-accounting discussion. I did not run tests or benchmarks.

Could you add a focused BASE/HEAD native-Celeborn microbenchmark for many partitions and constrained or slow workers, comparing throughput, enqueue/blocked time and retained bytes? Matched frame sizes, admission budgets and correctness checks would help quantify the new submission serialization and fallback concurrency limit.

[andygrove]

LGTM, no impact on non Celeborn users

[sunchao]

Thanks! merged