[SPARK-12617][PySpark]Clean up the leak sockets of Py4J

[zsxwing]

This patch added Py4jCallbackConnectionCleaner to clean the leak sockets of Py4J every 30 seconds. This is a workaround before Py4J fixes the leak issue py4j/py4j#187

[zsxwing]

CC @davies

[zsxwing]

This method is not called because PySpark never stops the gateway server.

[SparkQA]

Test build #48700 has finished for PR 10579 at commit bfd1b9a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #48691 has finished for PR 10579 at commit 950d5d0.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class Py4jCallbackConnectionCleaner(object):

[zsxwing]

retest this please

[SparkQA]

Test build #48710 has finished for PR 10579 at commit bfd1b9a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

LGTM

[davies]

Merged into master and 1.5, 1.6 branch

[davies]

Traceback (most recent call last):
File "//anaconda/lib/python2.7/threading.py", line 810, in __bootstrap_inner
self.run()
File "//anaconda/lib/python2.7/threading.py", line 1082, in run
self.function(_self.args, *_self.kwargs)
File "/Users/davies/work/spark/python/pyspark/context.py", line 78, in clean_closed_connections
with callback_server.lock:
AttributeError: 'NoneType' object has no attribute 'lock'

[zsxwing]

Sorry that I didn't notice that only Streaming uses the callback server. I sent #10621 to fix it.