[SPARK-22340][PYTHON][FOLLOW-UP] Add a better message and improve documentation for pinned thread mode

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

This PR proposes to show different warning message when the pinned thread mode is enabled:

When enabled:

> PYSPARK_PIN_THREAD feature is enabled. However, note that it cannot inherit the local properties from the parent thread although it isolates each thread on PVM and JVM with its own local properties.
> To work around this, you should manually copy and set the local properties from the parent thread to the child thread when you create another thread.

When disabled:

> Currently, 'setLocalProperty' (set to local properties) with multiple threads does not properly work.
> Internally threads on PVM and JVM are not synced, and JVM thread can be reused for multiple threads on PVM, which fails to isolate local properties for each thread on PVM.
> To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). However, note that it cannot inherit the local properties from the parent thread although it isolates each thread on PVM and JVM with its own local properties.
> To work around this, you should manually copy and set the local properties from the parent thread to the child thread when you create another thread.

### Why are the changes needed?

Currently, it shows the same warning message regardless of PYSPARK_PIN_THREAD being set. In the warning message it says "you can set PYSPARK_PIN_THREAD to true ..." which is confusing.

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

Documentation and warning message as shown above.

### How was this patch tested?

Manually tested.

```bash
$ PYSPARK_PIN_THREAD=true ./bin/pyspark
```

```python
sc.setJobGroup("a", "b")
```

```
.../pyspark/util.py:141: UserWarning: PYSPARK_PIN_THREAD feature is enabled. However, note that it cannot inherit the local properties from the parent thread although it isolates each thread on PVM and JVM with its own local properties.
To work around this, you should manually copy and set the local properties from the parent thread to the child thread when you create another thread.
  warnings.warn(msg, UserWarning)
```

```bash
$ ./bin/pyspark
```

```python
sc.setJobGroup("a", "b")
```

```
.../pyspark/util.py:141: UserWarning: Currently, 'setJobGroup' (set to local properties) with multiple threads does not properly work.
Internally threads on PVM and JVM are not synced, and JVM thread can be reused for multiple threads on PVM, which fails to isolate local properties for each thread on PVM.
To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). However, note that it cannot inherit the local properties from the parent thread although it isolates each thread on PVM and JVM with its own local properties.
To work around this, you should manually copy and set the local properties from the parent thread to the child thread when you create another thread.
  warnings.warn(msg, UserWarning)
```

Closes apache#26588 from HyukjinKwon/SPARK-22340.

Authored-by: HyukjinKwon <[email protected]>
Signed-off-by: HyukjinKwon <[email protected]>

Author: HyukjinKwon

python/pyspark/context.py

@@ -40,6 +40,7 @@
 from pyspark.traceback_utils import CallSite, first_spark_call
 from pyspark.status import StatusTracker
 from pyspark.profiler import ProfilerCollector, BasicProfiler
+from pyspark.util import _warn_pin_thread
 
 if sys.version > '3':
     xrange = range
@@ -1008,60 +1009,41 @@ def setJobGroup(self, groupId, description, interruptOnCancel=False):
         ensure that the tasks are actually stopped in a timely manner, but is off by default due
         to HDFS-1208, where HDFS may respond to Thread.interrupt() by marking nodes as dead.
 
-        .. note:: Currently, setting a group ID (set to local properties) with a thread does
-            not properly work. Internally threads on PVM and JVM are not synced, and JVM thread
-            can be reused for multiple threads on PVM, which fails to isolate local properties
-            for each thread on PVM. To work around this, you can set `PYSPARK_PIN_THREAD` to
+        .. note:: Currently, setting a group ID (set to local properties) with multiple threads
+            does not properly work. Internally threads on PVM and JVM are not synced, and JVM
+            thread can be reused for multiple threads on PVM, which fails to isolate local
+            properties for each thread on PVM.
+
+            To work around this, you can set `PYSPARK_PIN_THREAD` to
             `'true'` (see SPARK-22340). However, note that it cannot inherit the local properties
             from the parent thread although it isolates each thread on PVM and JVM with its own
-            local properties. To work around this, you should manually copy and set the local
+            local properties.
+
+            To work around this, you should manually copy and set the local
             properties from the parent thread to the child thread when you create another thread.
         """
-        warnings.warn(
-            "Currently, setting a group ID (set to local properties) with a thread does "
-            "not properly work. "
-            "\n"
-            "Internally threads on PVM and JVM are not synced, and JVM thread can be reused "
-            "for multiple threads on PVM, which fails to isolate local properties for each "
-            "thread on PVM. "
-            "\n"
-            "To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). "
-            "However, note that it cannot inherit the local properties from the parent thread "
-            "although it isolates each thread on PVM and JVM with its own local properties. "
-            "\n"
-            "To work around this, you should manually copy and set the local properties from "
-            "the parent thread to the child thread when you create another thread.",
-            UserWarning)
+        _warn_pin_thread("setJobGroup")
         self._jsc.setJobGroup(groupId, description, interruptOnCancel)
 
     def setLocalProperty(self, key, value):
         """
         Set a local property that affects jobs submitted from this thread, such as the
         Spark fair scheduler pool.
 
-        .. note:: Currently, setting a local property with a thread does
-            not properly work. Internally threads on PVM and JVM are not synced, and JVM thread
+        .. note:: Currently, setting a local property with multiple threads does not properly work.
+            Internally threads on PVM and JVM are not synced, and JVM thread
             can be reused for multiple threads on PVM, which fails to isolate local properties
-            for each thread on PVM. To work around this, you can set `PYSPARK_PIN_THREAD` to
+            for each thread on PVM.
+
+            To work around this, you can set `PYSPARK_PIN_THREAD` to
             `'true'` (see SPARK-22340). However, note that it cannot inherit the local properties
             from the parent thread although it isolates each thread on PVM and JVM with its own
-            local properties. To work around this, you should manually copy and set the local
+            local properties.
+
+            To work around this, you should manually copy and set the local
             properties from the parent thread to the child thread when you create another thread.
         """
-        warnings.warn(
-            "Currently, setting a local property with a thread does not properly work. "
-            "\n"
-            "Internally threads on PVM and JVM are not synced, and JVM thread can be reused "
-            "for multiple threads on PVM, which fails to isolate local properties for each "
-            "thread on PVM. "
-            "\n"
-            "To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). "
-            "However, note that it cannot inherit the local properties from the parent thread "
-            "although it isolates each thread on PVM and JVM with its own local properties. "
-            "\n"
-            "To work around this, you should manually copy and set the local properties from "
-            "the parent thread to the child thread when you create another thread.",
-            UserWarning)
+        _warn_pin_thread("setLocalProperty")
         self._jsc.setLocalProperty(key, value)
 
     def getLocalProperty(self, key):
@@ -1075,30 +1057,20 @@ def setJobDescription(self, value):
         """
         Set a human readable description of the current job.
 
-        .. note:: Currently, setting a job description (set to local properties) with a thread does
-            not properly work. Internally threads on PVM and JVM are not synced, and JVM thread
-            can be reused for multiple threads on PVM, which fails to isolate local properties
-            for each thread on PVM. To work around this, you can set `PYSPARK_PIN_THREAD` to
+        .. note:: Currently, setting a job description (set to local properties) with multiple
+            threads does not properly work. Internally threads on PVM and JVM are not synced,
+            and JVM thread can be reused for multiple threads on PVM, which fails to isolate
+            local properties for each thread on PVM.
+
+            To work around this, you can set `PYSPARK_PIN_THREAD` to
             `'true'` (see SPARK-22340). However, note that it cannot inherit the local properties
             from the parent thread although it isolates each thread on PVM and JVM with its own
-            local properties. To work around this, you should manually copy and set the local
+            local properties.
+
+            To work around this, you should manually copy and set the local
             properties from the parent thread to the child thread when you create another thread.
         """
-        warnings.warn(
-            "Currently, setting a job description (set to local properties) with a thread does "
-            "not properly work. "
-            "\n"
-            "Internally threads on PVM and JVM are not synced, and JVM thread can be reused "
-            "for multiple threads on PVM, which fails to isolate local properties for each "
-            "thread on PVM. "
-            "\n"
-            "To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). "
-            "However, note that it cannot inherit the local properties from the parent thread "
-            "although it isolates each thread on PVM and JVM with its own local properties. "
-            "\n"
-            "To work around this, you should manually copy and set the local properties from "
-            "the parent thread to the child thread when you create another thread.",
-            UserWarning)
+        _warn_pin_thread("setJobDescription")
         self._jsc.setJobDescription(value)
 
     def sparkUser(self):

python/pyspark/util.py

@@ -19,6 +19,8 @@
 import re
 import sys
 import traceback
+import os
+import warnings
 import inspect
 from py4j.protocol import Py4JJavaError
 
@@ -112,6 +114,33 @@ def wrapper(*args, **kwargs):
     return wrapper
 
 
+def _warn_pin_thread(name):
+    if os.environ.get("PYSPARK_PIN_THREAD", "false").lower() == "true":
+        msg = (
+            "PYSPARK_PIN_THREAD feature is enabled. "
+            "However, note that it cannot inherit the local properties from the parent thread "
+            "although it isolates each thread on PVM and JVM with its own local properties. "
+            "\n"
+            "To work around this, you should manually copy and set the local properties from "
+            "the parent thread to the child thread when you create another thread.")
+    else:
+        msg = (
+            "Currently, '%s' (set to local properties) with multiple threads does "
+            "not properly work. "
+            "\n"
+            "Internally threads on PVM and JVM are not synced, and JVM thread can be reused "
+            "for multiple threads on PVM, which fails to isolate local properties for each "
+            "thread on PVM. "
+            "\n"
+            "To work around this, you can set PYSPARK_PIN_THREAD to true (see SPARK-22340). "
+            "However, note that it cannot inherit the local properties from the parent thread "
+            "although it isolates each thread on PVM and JVM with its own local properties. "
+            "\n"
+            "To work around this, you should manually copy and set the local properties from "
+            "the parent thread to the child thread when you create another thread." % name)
+    warnings.warn(msg, UserWarning)
+
+
 def _print_missing_jar(lib_name, pkg_name, jar_name, spark_version):
     print("""
 ________________________________________________________________________________________________