fix: bind py4j callback server to a dynamic port to avoid 25334 collision (#86, #19)

[nikolauspschuetz]

Problem

Using a Check with a lambda assertion starts a py4j callback server. PythonCallback.__init__ called gateway.start_callback_server() with no port, so py4j binds the hardcoded default 25334. Concurrent or repeated runs on the same host then collide:

OSError: [Errno 98] Address already in use (127.0.0.1:25334)

This single root cause underlies a cluster of reports: #86, #19, #7, #72, #156, #173, #198. The known port=0 workaround alone is insufficient — the JVM-side callback client keeps pointing at the old port, producing Error while obtaining a new communication channel (#19).

Fix

Start the callback server on a dynamic port and reset the JVM-side callback client to the port actually bound — the documented py4j idiom (mirrors py4j's own ResetCallbackClientTest):

gateway.start_callback_server(CallbackServerParameters(port=0))
port = gateway.get_callback_server().get_listening_port()
gateway.java_gateway_server.resetCallbackClient(
    gateway.java_gateway_server.getCallbackClient().getAddress(), port)

The is_shutdown restart branch is also updated to re-derive a fresh dynamic port instead of reverting to 25334.

Verification

Runtime-verified on real Spark 3.5 / Java 17 / py4j 0.10.9.7 / Deequ 2.0.8:

• Lambda-assertion checks pass; the callback server binds a dynamic port (not 25334).
• Collision reproduction with port 25334 occupied by a blocking socket: the Check with assertion as parameter throws "OSError: [Errno 98] Address already in use" #86 path (bind collision) and the Why the callback server always listens to 25334 by default #19 path (second lambda check → "new communication channel") both succeed — ALL_SCENARIOS_PASSED.

Added test_lambda_check_uses_dynamic_callback_port to tests/test_checks.py, which squats on 25334, runs a lambda-assertion Check, and asserts the callback server does not use the default port.

Reviewer note

This touches the shared JVM-bridge callback lifecycle. The restart branch reaches into py4j's _callback_server attribute so get_callback_server() returns the new server — worth a careful look. getAddress() returns /127.0.0.1 (verified live) and is passed straight back into resetCallbackClient.

Closes #86, #19, #7, #72, #156, #173, #198

[github-actions]

---

Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: 416310f3) — may not be fully accurate. Reply if this doesn't help.

[nikolauspschuetz]

Ready for review. Reworked after deeper validation: the first approach (CallbackServerParameters(port=0) + resetCallbackClient) made shutdown_callback_server() hang at teardown on both Linux and macOS (I confirmed via the repo's own Dockerfile: stock shutdown returns cleanly, that approach hangs), which would have hung the test suite's tearDownClass in CI.

The current fix is minimal: set the gateway's existing (PySpark-configured) callback_server_parameters.port to 0 only if it's the legacy hardcoded 25334, then start the server the stock way. This avoids the collision and preserves PySpark's callback wiring, so shutdown returns cleanly. Verified on Linux (Docker) and macOS: full test_checks.py → 87 passed, exit 0, no teardown hang. cc @sudsali @chenliu0831. Closes #86, #19, #7, #72, #156, #173, #198.

[github-actions]

---

Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: 416310f3) — may not be fully accurate. Reply if this doesn't help.

[nikolauspschuetz]

Thanks for the automated review pass. The current revision addresses these findings:

• Restart path / is_shutdown — when the existing callback server is shut down, it is close()d before restart_callback_server(), so we never restart on a partially-torn-down server.
• No dependency on java_gateway_server — the fix mutates the gateway's existing callback_server_parameters (setting port = 0 only when it is the hardcoded DEFAULT_PYTHON_PROXY_PORT, 25334). This preserves PySpark's own callback wiring and clean shutdown; passing a fresh parameters object + resetCallbackClient hangs teardown on Linux/macOS, which is why that path is avoided.
• port = 0 is intentional and idempotent — port 0 asks the OS for a free ephemeral port on each bind, which is exactly the collision fix (Check with assertion as parameter throws "OSError: [Errno 98] Address already in use" #86/Why the callback server always listens to 25334 by default #19/spark-submit job is not exiting until I hit ctrl+C if I use Check class #7/Example never ends because of some deequ's alive thread #72/AWS Glue Job runs indefinitely after exception occurs when adding Checks to the VerificationSuite #156/Running multiple pydeequ checks at the same time in emr cluster OSError: [Errno 98] Address already in use :(127.0.0.1:25334)  #173/Issue with application exit while using pydeequ #198). It is only set when the param is currently the 25334 default.
• Test isolation — the dynamic-port behavior is exercised by forcing the callback_server is None branch; the shared Spark session is managed in setUp/tearDown.

Resolving the threads accordingly.