[SPARK-54619][PYTHON] Add a sanity check for configuration numbers

### What changes were proposed in this pull request?

Add sanity check for number of configurations being passed.

### Why are the changes needed?

This is helpful to recognize malformed message - avoid potential deadlock when the message does not conform to protocol.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

This error should not happen and it should not break CI either.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #53359 from gaogaotiantian/runner-conf-sanity-check.

Authored-by: Tian Gao <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: gaogaotiantian

python/pyspark/errors/error-conditions.json

@@ -941,6 +941,11 @@
       "Argument <arg_name> must be a numerical column for plotting, got <arg_type>."
     ]
   },
+  "PROTOCOL_ERROR": {
+    "message": [
+      "<failure>. This usually indicates that the message does not conform to the protocol."
+    ]
+  },
   "PYTHON_HASH_SEED_NOT_SET": {
     "message": [
       "Randomness of hash of string should be disabled via PYTHONHASHSEED."

python/pyspark/worker.py

@@ -118,7 +118,19 @@ def __init__(self, infile=None):
 
     def load(self, infile):
         num_conf = read_int(infile)
-        for i in range(num_conf):
+        # We do a sanity check here to reduce the possibility to stuck indefinitely
+        # due to an invalid messsage. If the numer of configurations is obviously
+        # wrong, we just raise an error directly.
+        # We hand-pick the configurations to send to the worker so the number should
+        # be very small (less than 100).
+        if num_conf < 0 or num_conf > 10000:
+            raise PySparkRuntimeError(
+                errorClass="PROTOCOL_ERROR",
+                messageParameters={
+                    "failure": f"Invalid number of configurations: {num_conf}",
+                },
+            )
+        for _ in range(num_conf):
             k = utf8_deserializer.loads(infile)
             v = utf8_deserializer.loads(infile)
             self._conf[k] = v