[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.
Even without this assumption, There is only 1/2^32 chance to misunderstanding the next 4 bytes. I think it's still acceptable in practice.