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:
[email protected]
