[GitHub] [flink] dianfu commented on a change in pull request #12061: [FLINK-17454][python]python gateway callback server port should not b…

[GitBox <gi...@apache.org>]

dianfu commented on a change in pull request #12061:
URL: https://github.com/apache/flink/pull/12061#discussion_r422746320



##########
File path: flink-python/src/main/java/org/apache/flink/client/python/PythonEnvUtils.java
##########
@@ -300,6 +294,48 @@ static GatewayServer startGatewayServer() throws ExecutionException, Interrupted
 		return gatewayServerFuture.get();
 	}
 
+	/**
+	 * Reset a daemon thread to the callback client thread pool so that the callback server can be terminated when gate
+	 * way server is shutting down. We need to shut down the none-daemon thread firstly, then set a new thread created
+	 * in a daemon thread to the ExecutorService.
+	 *
+	 * @param gatewayServer the gateway which creates the callback server.
+	 * */
+	private static void resetCallbackClientExecutorService(GatewayServer gatewayServer) throws NoSuchFieldException,
+		IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+		CallbackClient callbackClient = (CallbackClient) gatewayServer.getCallbackClient();
+		// The Java API of py4j does not provide approach to set "daemonize_connections" parameter.
+		// Use reflect to daemonize the connection thread.
+		Field executor = CallbackClient.class.getDeclaredField("executor");
+		executor.setAccessible(true);
+		((ScheduledExecutorService) executor.get(callbackClient)).shutdown();
+		executor.set(callbackClient, Executors.newScheduledThreadPool(1, Thread::new));
+		Method setupCleaner = CallbackClient.class.getDeclaredMethod("setupCleaner");
+		setupCleaner.setAccessible(true);
+		setupCleaner.invoke(callbackClient);
+	}
+
+	/**
+	 * Reset the callback client of gatewayServer with the given callbackListeningAddress and callbackListeningPort
+	 * after the callback server started.
+	 *
+	 * @param callbackListeningAddress the listening address of the callback server.
+	 * @param callbackListeningPort the listening port of the callback server.
+	 * */
+	public static void resetCallbackClient(String callbackListeningAddress, int callbackListeningPort) throws
+		UnknownHostException, InvocationTargetException, NoSuchMethodException, IllegalAccessException,
+		NoSuchFieldException {
+
+		gatewayServer = getGatewayServer();
+		if (gatewayServer == null){

Review comment:
       It seems that it is never `null`.

##########
File path: flink-python/pyflink/java_gateway.py
##########
@@ -49,15 +49,19 @@ def get_gateway():
             # if Java Gateway is already running
             if 'PYFLINK_GATEWAY_PORT' in os.environ:
                 gateway_port = int(os.environ['PYFLINK_GATEWAY_PORT'])
-                callback_port = int(os.environ['PYFLINK_CALLBACK_PORT'])
                 gateway_param = GatewayParameters(port=gateway_port, auto_convert=True)
                 _gateway = JavaGateway(
                     gateway_parameters=gateway_param,
                     callback_server_parameters=CallbackServerParameters(
-                        port=callback_port, daemonize=True, daemonize_connections=True))
+                        port=0, daemonize=True, daemonize_connections=True))
             else:
                 _gateway = launch_gateway()
 
+            callback_server = _gateway.get_callback_server()
+            listening_callback_address = callback_server.get_listening_address()

Review comment:
       rename to `callback_server_listening_address` and `callback_server_listening_port `




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org