docstr: Starts the Spark Connect server given the master (thread-unsafe).
  
  At the high level, there are two cases. The first case is development case, e.g.,
  you locally build Apache Spark, and run ``SparkSession.builder.remote("local")``:
  
  1. This method automatically finds the jars for Spark Connect (because the jars for
    Spark Connect are not bundled in the regular Apache Spark release).
  
  2. Temporarily remove all states for Spark Connect, for example, ``SPARK_REMOTE``
    environment variable.
  
  3. Starts a JVM (without Spark Context) first, and adds the Spark Connect server jars
     into the current class loader. Otherwise, Spark Context with ``spark.plugins``
     cannot be initialized because the JVM is already running without the jars in
     the classpath before executing this Python process for driver side (in case of
     PySpark application submission).
  
  4. Starts a regular Spark session that automatically starts a Spark Connect server
     via ``spark.plugins`` feature.
  
  The second case is when you use Apache Spark release:
  
  1. Users must specify either the jars or package, e.g., ``--packages
    org.apache.spark:spark-connect_2.12:3.4.0``. The jars or packages would be specified
    in SparkSubmit automatically. This method does not do anything related to this.
  
  2. Temporarily remove all states for Spark Connect, for example, ``SPARK_REMOTE``
    environment variable. It does not do anything for PySpark application submission as
    well because jars or packages were already specified before executing this Python
    process for driver side.
  
  3. Starts a regular Spark session that automatically starts a Spark Connect server
    with JVM via ``spark.plugins`` feature.
docstr_tok: ["Starts","the","Spark","Connect","server","given","the","master","(","thread-unsafe",")",".","At","the","high","level",",","there","are","two","cases",".","The","first","case","is","development","case",",","e.g.",",","you","locally","build","Apache","Spark",",","and","run","``","SparkSession.builder.remote","(","``","local","''",")","``",":","1",".","This","method","automatically","finds","the","jars","for","Spark","Connect","(","because","the","jars","for","Spark","Connect","are","not","bundled","in","the","regular","Apache","Spark","release",")",".","2",".","Temporarily","remove","all","states","for","Spark","Connect",",","for","example",",","``","SPARK_REMOTE","``","environment","variable",".","3",".","Starts","a","JVM","(","without","Spark","Context",")","first",",","and","adds","the","Spark","Connect","server","jars","into","the","current","class","loader",".","Otherwise",",","Spark","Context","with","``","spark.plugins","``","can","not","be","initialized","because","the","JVM","is","already","running","without","the","jars","in","the","classpath","before","executing","this","Python","process","for","driver","side","(","in","case","of","PySpark","application","submission",")",".","4",".","Starts","a","regular","Spark","session","that","automatically","starts","a","Spark","Connect","server","via","``","spark.plugins","``","feature",".","The","second","case","is","when","you","use","Apache","Spark","release",":","1",".","Users","must","specify","either","the","jars","or","package",",","e.g.",",","``","--","packages","org.apache.spark",":","spark-connect_2.12:3.4.0","``",".","The","jars","or","packages","would","be","specified","in","SparkSubmit","automatically",".","This","method","does","not","do","anything","related","to","this",".","2",".","Temporarily","remove","all","states","for","Spark","Connect",",","for","example",",","``","SPARK_REMOTE","``","environment","variable",".","It","does","not","do","anything","for","PySpark","application","submission","as","well","because","jars","or","packages","were","already","specified","before","executing","this","Python","process","for","driver","side",".","3",".","Starts","a","regular","Spark","session","that","automatically","starts","a","Spark","Connect","server","with","JVM","via","``","spark.plugins","``","feature","."]
code: def _start_connect_server(master: str, opts: Dict[str, Any]) -> None:
      """
      Starts the Spark Connect server given the master (thread-unsafe).
  
      At the high level, there are two cases. The first case is development case, e.g.,
      you locally build Apache Spark, and run ``SparkSession.builder.remote("local")``:
  
      1. This method automatically finds the jars for Spark Connect (because the jars for
        Spark Connect are not bundled in the regular Apache Spark release).
  
      2. Temporarily remove all states for Spark Connect, for example, ``SPARK_REMOTE``
        environment variable.
  
      3. Starts a JVM (without Spark Context) first, and adds the Spark Connect server jars
         into the current class loader. Otherwise, Spark Context with ``spark.plugins``
         cannot be initialized because the JVM is already running without the jars in
         the classpath before executing this Python process for driver side (in case of
         PySpark application submission).
  
      4. Starts a regular Spark session that automatically starts a Spark Connect server
         via ``spark.plugins`` feature.
  
      The second case is when you use Apache Spark release:
  
      1. Users must specify either the jars or package, e.g., ``--packages
        org.apache.spark:spark-connect_2.12:3.4.0``. The jars or packages would be specified
        in SparkSubmit automatically. This method does not do anything related to this.
  
      2. Temporarily remove all states for Spark Connect, for example, ``SPARK_REMOTE``
        environment variable. It does not do anything for PySpark application submission as
        well because jars or packages were already specified before executing this Python
        process for driver side.
  
      3. Starts a regular Spark session that automatically starts a Spark Connect server
        with JVM via ``spark.plugins`` feature.
      """
      session = PySparkSession._instantiatedSession
      if session is None or session._sc._jsc is None:
          # Configurations to be overwritten
          overwrite_conf = opts
          overwrite_conf["spark.master"] = master
          overwrite_conf["spark.local.connect"] = "1"
  
          # Configurations to be set if unset.
          default_conf = {
              "spark.plugins": "org.apache.spark.sql.connect.SparkConnectPlugin"
          }
  
          if "SPARK_TESTING" in os.environ:
              # For testing, we use 0 to use an ephemeral port to allow parallel testing.
              # See also SPARK-42272.
              overwrite_conf["spark.connect.grpc.binding.port"] = "0"
  
          def create_conf(**kwargs: Any) -> SparkConf:
              conf = SparkConf(**kwargs)
              for k, v in overwrite_conf.items():
                  conf.set(k, v)
              for k, v in default_conf.items():
                  if not conf.contains(k):
                      conf.set(k, v)
              return conf
  
          # Check if we're using unreleased version that is in development.
          # Also checks SPARK_TESTING for RC versions.
          is_dev_mode = (
              "dev" in LooseVersion(__version__).version
              or "SPARK_TESTING" in os.environ
          )
  
          origin_remote = os.environ.get("SPARK_REMOTE", None)
          try:
              if origin_remote is not None:
                  # So SparkSubmit thinks no remote is set in order to
                  # start the regular PySpark session.
                  del os.environ["SPARK_REMOTE"]
  
              SparkContext._ensure_initialized(
                  conf=create_conf(loadDefaults=False)
              )
  
              if is_dev_mode:
                  # Try and catch for a possibility in production because pyspark.testing
                  # does not exist in the canonical release.
                  try:
                      from pyspark.testing.utils import search_jar
  
                      # Note that, in production, spark.jars.packages configuration should be
                      # set by users. Here we're automatically searching the jars locally built.
                      connect_jar = search_jar(
                          "connector/connect/server",
                          "spark-connect-assembly-",
                          "spark-connect",
                      )
                      if connect_jar is None:
                          warnings.warn(
                              "Attempted to automatically find the Spark Connect jars because "
                              "'SPARK_TESTING' environment variable is set, or the current "
                              f"PySpark version is dev version ({__version__}). However, the jar"
                              " was not found. Manually locate the jars and specify them, e.g., "
                              "'spark.jars' configuration."
                          )
                      else:
                          pyutils = SparkContext._jvm.PythonSQLUtils  # type: ignore[union-attr]
                          pyutils.addJarToCurrentClassLoader(
                              connect_jar
                          )
  
                          # Required for local-cluster testing as their executors need the jars
                          # to load the Spark plugin for Spark Connect.
                          if master.startswith("local-cluster"):
                              if "spark.jars" in overwrite_conf:
                                  overwrite_conf[
                                      "spark.jars"
                                  ] = f"{overwrite_conf['spark.jars']},{connect_jar}"
                              else:
                                  overwrite_conf[
                                      "spark.jars"
                                  ] = connect_jar
  
                  except ImportError:
                      pass
  
              # The regular PySpark session is registered as an active session
              # so would not be garbage-collected.
              PySparkSession(
                  SparkContext.getOrCreate(
                      create_conf(
                          loadDefaults=True, _jvm=SparkContext._jvm
                      )
                  )
              )
  
              # Lastly only keep runtime configurations because other configurations are
              # disallowed to set in the regular Spark Connect session.
              utl = SparkContext._jvm.PythonSQLUtils  # type: ignore[union-attr]
              runtime_conf_keys = [
                  c._1() for c in utl.listRuntimeSQLConfigs()
              ]
              new_opts = {
                  k: opts[k] for k in opts if k in runtime_conf_keys
              }
              opts.clear()
              opts.update(new_opts)
  
          finally:
              if origin_remote is not None:
                  os.environ["SPARK_REMOTE"] = origin_remote
      else:
          raise PySparkRuntimeError(
              error_class="SESSION_OR_CONTEXT_EXISTS",
              message_parameters={},
          )
  
imports: from pyspark.sql.connect.utils import check_dependencies
  import threading
  import os
  import warnings
  from collections.abc import Sized
  from functools import reduce
  from threading import RLock
  from typing import Optional, Any, Union, Dict, List, Tuple, Set, cast, overload, Iterable, TYPE_CHECKING, ClassVar
  import numpy
  import pandas
  import pyarrow
  from pandas.api.types import is_datetime64_dtype, is_timedelta64_dtype
  import urllib
  from pyspark import SparkContext, SparkConf, __version__
  from pyspark.loose_version import LooseVersion
  from pyspark.sql.connect.client import SparkConnectClient, ChannelBuilder
  from pyspark.sql.connect.conf import RuntimeConf
  from pyspark.sql.connect.dataframe import DataFrame
  from pyspark.sql.connect.plan import SQL, Range, LocalRelation, LogicalPlan, CachedLocalRelation, CachedRelation, CachedRemoteRelation
  from pyspark.sql.connect.readwriter import DataFrameReader
  from pyspark.sql.connect.streaming.readwriter import DataStreamReader
  from pyspark.sql.connect.streaming.query import StreamingQueryManager
  from pyspark.sql.pandas.serializers import ArrowStreamPandasSerializer
  from pyspark.sql.pandas.types import to_arrow_schema, to_arrow_type, _deduplicate_field_names
  from pyspark.sql.session import classproperty, SparkSession
  from pyspark.sql.types import _infer_schema, _has_nulltype, _merge_type, Row, DataType, DayTimeIntervalType, StructType, AtomicType, TimestampType
  from pyspark.sql.utils import to_str
  from pyspark.errors import PySparkAttributeError, PySparkNotImplementedError, PySparkRuntimeError, PySparkValueError, PySparkTypeError
label_desc: Use image node_id 35 for calling the SparkSession obj's underlying member method code with example usage: obj._start_connect_server(master, opts) and returns: conf