[SPARK-3993] [PySpark] fix bug while reuse worker after take()

After take(), maybe there are some garbage left in the socket, then next task assigned to this worker will hang because of corrupted data.

We should make sure the socket is clean before reuse it, write END_OF_STREAM at the end, and check it after read out all result from python.
The main() is called in a loop after re-use patch, so 'return' will not let the worker exit, this break the assumption that hold before the patch.

A better approach may be remove this special cases, and rewrite the test cases without any special treatments.