How to use the py4j.java_gateway.CallbackServerParameters function in py4j

  @classmethod
  def __ensureGatewayInit(cls, gatewayPort, driver):
    with cls._lock:
      if not cls._gateway:
        # Spark 1.6 and Spark 2 are using later verions of py4j (0.9 and 0.10+ respectively),
        # which has better control on gateway client and callback server using
        # GatewayParameters and CallbackServerParameters. Try to use those,
        # as it'll be less hacky (it's still a bit hacky for Spark 1.6, see below)
        try:
          from py4j.java_gateway import GatewayParameters, CallbackServerParameters
          callbackServerParams = CallbackServerParameters(port = 0, daemonize = True,
                                                          daemonize_connections = True) if driver else None
          gateway = JavaGateway(gateway_parameters = GatewayParameters(port = gatewayPort, auto_convert = True),
                                callback_server_parameters = callbackServerParams)
        except:
          from py4j.java_gateway import CallbackServer, GatewayClient
          gateway = JavaGateway(gateway_client = GatewayClient(port = gatewayPort), auto_convert = True)
          cls._onDemandCallback = True

        java_import(gateway.jvm, "io.cdap.cdap.app.runtime.spark.*")
        java_import(gateway.jvm, "io.cdap.cdap.app.runtime.spark.python.*")

        if driver and not cls._onDemandCallback:
          # For py4j 0.10+ (used by Spark 2.0), use the official API to set set callback port on the gateway server
          if "get_callback_server" in dir(gateway):
            callbackPort = gateway.get_callback_server().get_listening_port()
            gateway.jvm.SparkPythonUtil.setGatewayCallbackPort(gateway.java_gateway_server, callbackPort)

def _start_gateway(self):
        # auto select callback server port and reset it in java env
        self.gateway = JavaGateway(gateway_parameters=GatewayParameters(
            port=self.port), callback_server_parameters=CallbackServerParameters(port=0))
        python_port = self.gateway.get_callback_server().get_listening_port()
        self.gateway.java_gateway_server.resetCallbackClient(
            self.gateway.java_gateway_server.getCallbackClient().getAddress(), python_port)
        self.manager = self.gateway.entry_point

        # check if pipe built
        if self.p1_server is None:
            raise Exception("Must call build_pipe_and_return_p2 and also make p2 env after gym.make() but before env.reset()")
        self.pipe = self.p1_server

        if self.display:
            self.p1 = GymAIDisplay(self.gateway, self.p1_client, self.frameskip)
            self.p2 = GymAIDisplay(self.gateway, self.p2_client, self.frameskip)
        else:
            self.p1 = GymAI(self.gateway, self.p1_client, self.frameskip)
            self.p2 = GymAI(self.gateway, self.p2_client, self.frameskip)

def _setup_py4j_client_connection(self):
        gateway_params = GatewayParameters(port=self._java_port,
                                           auto_field=True,
                                           auto_close=True,
                                           eager_load=True)
        callback_server_params = CallbackServerParameters(port=0,
                                                          daemonize=True,
                                                          daemonize_connections=True,
                                                          eager_load=True)
        self._gateway = JavaGateway(gateway_parameters=gateway_params,
                                    callback_server_parameters=callback_server_params,
                                    python_server_entry_point=self)
        self._component_via_py4j = self._gateway.entry_point.getComponent()
        if not self._component_via_py4j:
            raise MLCompException("None reference of py4j java object!")

        if self._verbose:
            self._logger.debug(self._prefix_msg + "Py4J component referenced successfully! comp_via_py4j: {}"
                               .format(self._component_via_py4j))

        self._component_via_py4j.setEnvAttributes(self.get_wid(), self._verbose)

def _validate(self, _):
        # here is where we can get java and python ports and change the
        # defaults for connecting
        try:
            self._bridge = Py4jServiceBridge(
                service_listener=self,
                gateway_parameters=GatewayParameters(port=self._javaport),
                callback_server_parameters=CallbackServerParameters(
                    port=self._pythonport
                ),
            )
            self._bridge.connect()
        except Exception as e:
            self._bridge = None
            raise e
        # Once bridge is connected, instantiate container using bridge id
        container_props = self._prepare_container_props(
            self._supported_intents, None
        )
        if self._default_service_timeout:
            container_props[
                ECF_PY4J_DEFAULT_SERVICE_TIMEOUT
            ] = self._default_service_timeout
        self._container = self._ipopo.instantiate(

port = launch_gateway(port=port,
                              classpath=Context.find_resource_path('net-0.8.56.jar'),
                              javaopts=["--add-modules", "java.se",
                                        "--add-exports", "java.base/jdk.internal.ref=ALL-UNNAMED",
                                        "--add-opens", "java.base/java.lang=ALL-UNNAMED",
                                        "--add-opens", "java.base/java.nio=ALL-UNNAMED",
                                        "--add-opens", "java.base/sun.nio.ch=ALL-UNNAMED",
                                        "--add-opens", "java.management/sun.management=ALL-UNNAMED",
                                        "--add-opens", "jdk.management/com.sun.management.internal=ALL-UNNAMED"],
                              die_on_exit=True)

        # connect python side to Java side with Java dynamic port and start python
        # callback server with a dynamic port
        gateway = JavaGateway(
            gateway_parameters=GatewayParameters(port=port, auto_convert=True),
            callback_server_parameters=CallbackServerParameters(port=0))

        # retrieve the port on which the python callback server was bound to.
        python_port = gateway.get_callback_server().get_listening_port()

        # tell the Java side to connect to the python callback server with the new
        # python port. Note that we use the java_gateway_server attribute that
        # retrieves the GatewayServer instance.
        gateway.java_gateway_server.resetCallbackClient(
            gateway.java_gateway_server.getCallbackClient().getAddress(),
            python_port)
        return gateway

def _start_gateway(self, p2=Machete):
        # auto select callback server port and reset it in java env
        self.gateway = JavaGateway(gateway_parameters=GatewayParameters(
            port=self.port), callback_server_parameters=CallbackServerParameters(port=0))
        python_port = self.gateway.get_callback_server().get_listening_port()
        self.gateway.java_gateway_server.resetCallbackClient(
            self.gateway.java_gateway_server.getCallbackClient().getAddress(), python_port)
        self.manager = self.gateway.entry_point

        # create pipe between gym_env_api and python_ai for java env
        server, client = Pipe()
        self.pipe = server
        self.p1 = GymAI(self.gateway, client, True)
        self.manager.registerAI(self.p1.__class__.__name__, self.p1)

        if isinstance(p2, str):
            # p2 is a java class name
            self.p2 = p2
            self.game_to_start = self.manager.createGame(
                "ZEN", "ZEN", self.p1.__class__.__name__, self.p2, self.freq_restart_java)

def connect(self,callback_server_parameters=None):
        if not callback_server_parameters:
            callback_server_parameters = CallbackServerParameters()
        with self._lock:
            if not self._gateway is None:
                raise ConnectionError('already connected to java gateway')
            server_started.connect(self._started)
            self._gateway = JavaGateway(callback_server_parameters=callback_server_parameters)
            cbserver = self._gateway.get_callback_server()
            server_stopped.connect(
                self._stopped, sender=cbserver)
            server_connection_started.connect(
                self._connection_started,
                sender=cbserver)
            server_connection_stopped.connect(
                self._connection_stopped,
                sender=cbserver)
            pre_server_shutdown.connect(
                self._pre_shutdown, sender=cbserver)

def _start_gateway(self, p2=Machete):
        # auto select callback server port and reset it in java env
        self.gateway = JavaGateway(gateway_parameters=GatewayParameters(
            port=self.port), callback_server_parameters=CallbackServerParameters(port=0))
        python_port = self.gateway.get_callback_server().get_listening_port()
        self.gateway.java_gateway_server.resetCallbackClient(
            self.gateway.java_gateway_server.getCallbackClient().getAddress(), python_port)
        self.manager = self.gateway.entry_point

        # create pipe between gym_env_api and python_ai for java env
        server, client = Pipe()
        self.pipe = server
        self.p1 = GymAI(self.gateway, client, False)
        self.manager.registerAI(self.p1.__class__.__name__, self.p1)

        if isinstance(p2, str):
            # p2 is a java class name
            self.p2 = p2
            self.game_to_start = self.manager.createGame(
                "ZEN", "ZEN", self.p1.__class__.__name__, self.p2, self.freq_restart_java)

the garbage collection requests will be daemonized. This means that
            the Python side might not send all garbage collection requests if
            it exits. If False, memory management will block the Python program
            exit until all requests are sent.

        :param auth_token: if provided, an authentication that token clients
            must provide to the server when connecting.
        """
        super(JavaParameters, self).__init__(
            address, port, auto_field, auto_close, auto_convert, eager_load,
            ssl_context, enable_memory_management, read_timeout, auth_token)
        self.auto_gc = auto_gc
        self.daemonize_memory_management = daemonize_memory_management


class PythonParameters(CallbackServerParameters):
    """Wrapper class that contains all parameters that can be passed to
    configure a `ClientServer`
    """

    def __init__(
            self, address=DEFAULT_ADDRESS, port=DEFAULT_PYTHON_PROXY_PORT,
            daemonize=False, daemonize_connections=False, eager_load=True,
            ssl_context=None, auto_gc=False,
            accept_timeout=DEFAULT_ACCEPT_TIMEOUT_PLACEHOLDER,
            read_timeout=None, propagate_java_exceptions=False,
            auth_token=None):
        """
        :param address: the address to which the client will request a
            connection

        :param port: the port to which the client will request a connection.