[SPARK-35834][PYTHON] Use the same cleanup logic as Py4J in inheritable thread API
[SPARK-35834][PYTHON] Use the same cleanup logic as Py4J in inheritable thread API

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

This PR fixes the cleanup logic in inheritable thread API by following Py4J cleanup logic at https://github.com/bartdag/py4j/blob/master/py4j-python/src/py4j/clientserver.py#L269-L278.

Currently the tests that use `inheritable_thread_target` are flaky (https://github.com/apache/spark/runs/2870944288):

```
======================================================================
ERROR [71.813s]: test_save_load_pipeline_estimator (pyspark.ml.tests.test_tuning.CrossValidatorTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/__w/spark/spark/python/pyspark/ml/tests/test_tuning.py", line 589, in test_save_load_pipeline_estimator
    self._run_test_save_load_pipeline_estimator(DummyLogisticRegression)
  File "/__w/spark/spark/python/pyspark/ml/tests/test_tuning.py", line 572, in _run_test_save_load_pipeline_estimator
    cvModel2 = crossval2.fit(training)
  File "/__w/spark/spark/python/pyspark/ml/base.py", line 161, in fit
    return self._fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/tuning.py", line 747, in _fit
    bestModel = est.fit(dataset, epm[bestIndex])
  File "/__w/spark/spark/python/pyspark/ml/base.py", line 159, in fit
    return self.copy(params)._fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/pipeline.py", line 114, in _fit
    model = stage.fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/base.py", line 161, in fit
    return self._fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/pipeline.py", line 114, in _fit
    model = stage.fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/base.py", line 161, in fit
    return self._fit(dataset)
  File "/__w/spark/spark/python/pyspark/ml/classification.py", line 2924, in _fit
    models = pool.map(inheritable_thread_target(trainSingleClass), range(numClasses))
  File "/__t/Python/3.6.13/x64/lib/python3.6/multiprocessing/pool.py", line 266, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/__t/Python/3.6.13/x64/lib/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
  File "/__t/Python/3.6.13/x64/lib/python3.6/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/__t/Python/3.6.13/x64/lib/python3.6/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))
  File "/__w/spark/spark/python/pyspark/util.py", line 324, in wrapped
    InheritableThread._clean_py4j_conn_for_current_thread()
  File "/__w/spark/spark/python/pyspark/util.py", line 389, in _clean_py4j_conn_for_current_thread
    del connections[i]
IndexError: deque index out of range

----------------------------------------------------------------------
```

This seems to be because the connection deque `jvm._gateway_client.deque` is accessed, and modified by other threads. Therefore, the number of threads could be changed in the middle. Using `SparkContext._lock` doesn't protect because the deque can be updated for every Java instance access in Py4J.

This PR proposes to use the atomic `deque.remove` in the problematic dequeue alone with try-catch on `ValueError` in case it's [deleted by Py4J](https://github.com/bartdag/py4j/blob/master/py4j-python/src/py4j/clientserver.py#L269-L278).

### Why are the changes needed?

To fix the flakiness in the tests, and avoid possible breakage in user application by using this API.

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

If users were dependent on InheritableThread with pinned thread mode on, they might have faced such issues intermittently. This PR fixes it.

### How was this patch tested?

Manually tested. CI should test it out too.

Closes #32989 from HyukjinKwon/SPARK-35834.

Authored-by: Hyukjin Kwon <2680965ad5aae29275c8fb10150cb10f5554e78b@apache.org>
Signed-off-by: Hyukjin Kwon <2680965ad5aae29275c8fb10150cb10f5554e78b@apache.org>

WeichenXu123/spark,shaneknapp/spark,taroplus/spark,zero323/spark,WeichenXu123/spark,ueshin/apache-spark,shaneknapp/spark,maropu/spark,holdenk/spark,xuanyuanking/spark,holdenk/spark,chuckchen/spark,apache/spark,gengliangwang/spark,zzcclp/spark,ueshin/apache-spark,apache/spark,chuckchen/spark,holdenk/spark,jiangxb1987/spark,rednaxelafx/apache-spark,chuckchen/spark,zero323/spark,maropu/spark,wangyum/spark,xuanyuanking/spark,WeichenXu123/spark,rednaxelafx/apache-spark,HyukjinKwon/spark,srowen/spark,milliman/spark,shaneknapp/spark,HyukjinKwon/spark,milliman/spark,taroplus/spark,chuckchen/spark,shaneknapp/spark,HyukjinKwon/spark,holdenk/spark,hvanhovell/spark,zzcclp/spark,apache/spark,HyukjinKwon/spark,rednaxelafx/apache-spark,zero323/spark,cloud-fan/spark,dongjoon-hyun/spark,jiangxb1987/spark,nchammas/spark,milliman/spark,HyukjinKwon/spark,shaneknapp/spark,milliman/spark,dongjoon-hyun/spark,dongjoon-hyun/spark,rednaxelafx/apache-spark,milliman/spark,zero323/spark,mahak/spark,srowen/spark,srowen/spark,chuckchen/spark,maropu/spark,mahak/spark,taroplus/spark,jiangxb1987/spark,wangyum/spark,maropu/spark,ueshin/apache-spark,witgo/spark,mahak/spark,wangmiao1981/spark,srowen/spark,xuanyuanking/spark,JoshRosen/spark,wangmiao1981/spark,JoshRosen/spark,taroplus/spark,JoshRosen/spark,WeichenXu123/spark,vinodkc/spark,maropu/spark,zzcclp/spark,wangmiao1981/spark,apache/spark,cloud-fan/spark,zzcclp/spark,rednaxelafx/apache-spark,rednaxelafx/apache-spark,hvanhovell/spark,vinodkc/spark,vinodkc/spark,nchammas/spark,dongjoon-hyun/spark,gengliangwang/spark,cloud-fan/spark,nchammas/spark,chuckchen/spark,chuckchen/spark,wangmiao1981/spark,jiangxb1987/spark,dongjoon-hyun/spark,jiangxb1987/spark,cloud-fan/spark,xuanyuanking/spark,wangyum/spark,xuanyuanking/spark,dongjoon-hyun/spark,cloud-fan/spark,nchammas/spark,taroplus/spark,gengliangwang/spark,mahak/spark,zzcclp/spark,WeichenXu123/spark,taroplus/spark,HyukjinKwon/spark,zero323/spark,hvanhovell/spark,witgo/spark,xuanyuanking/spark,wangyum/spark,witgo/spark,wangmiao1981/spark,nchammas/spark,witgo/spark,dongjoon-hyun/spark,milliman/spark,mahak/spark,mahak/spark,cloud-fan/spark,holdenk/spark,apache/spark,gengliangwang/spark,srowen/spark,apache/spark,mahak/spark,holdenk/spark,vinodkc/spark,WeichenXu123/spark,zzcclp/spark,jiangxb1987/spark,zzcclp/spark,srowen/spark,WeichenXu123/spark,hvanhovell/spark,apache/spark,HyukjinKwon/spark,hvanhovell/spark,rednaxelafx/apache-spark,maropu/spark,shaneknapp/spark,wangyum/spark,jiangxb1987/spark,srowen/spark,vinodkc/spark,zero323/spark,ueshin/apache-spark,milliman/spark,shaneknapp/spark,vinodkc/spark,JoshRosen/spark,wangmiao1981/spark,xuanyuanking/spark,witgo/spark,ueshin/apache-spark,maropu/spark,holdenk/spark,JoshRosen/spark,zero323/spark,witgo/spark,gengliangwang/spark,taroplus/spark,wangyum/spark,hvanhovell/spark,JoshRosen/spark,nchammas/spark,vinodkc/spark,nchammas/spark,cloud-fan/spark,ueshin/apache-spark,witgo/spark,gengliangwang/spark,wangmiao1981/spark,gengliangwang/spark,ueshin/apache-spark,JoshRosen/spark,hvanhovell/spark,wangyum/spark