[FLINK-17945][python] Improve the error message when instantiating no…

…n-existing Java class

This closes apache#12469.

flink-python/pyflink/java_gateway.py

@@ -28,7 +28,7 @@
 from py4j.java_gateway import (java_import, logger, JavaGateway, GatewayParameters,
                                CallbackServerParameters)
 from pyflink.pyflink_gateway_server import launch_gateway_server_process
-from pyflink.util.exceptions import install_exception_handler
+from pyflink.util.exceptions import install_exception_handler, install_py4j_hooks
 
 _gateway = None
 _lock = RLock()
@@ -69,6 +69,7 @@ def get_gateway():
             # import the flink view
             import_flink_view(_gateway)
             install_exception_handler()
+            install_py4j_hooks()
             _gateway.entry_point.put("PythonFunctionFactory", PythonFunctionFactory())
             _gateway.entry_point.put("Watchdog", Watchdog())
     return _gateway

flink-python/pyflink/util/exceptions.py

@@ -172,3 +172,15 @@ def install_exception_handler():
     patched = capture_java_exception(original)
     # only patch the one used in py4j.java_gateway (call Java API)
     py4j.java_gateway.get_return_value = patched
+
+
+def install_py4j_hooks():
+    """
+    Hook the classes such as JavaPackage, etc of Py4j to improve the exception message.
+    """
+    def wrapped_call(self, *args, **kwargs):
+        raise TypeError(
+            "Could not found the Java class '%s'. The Java dependencies could be specified via "
+            "command line argument '--jarfile' or the config option 'pipeline.jars'" % self._fqn)
+
+    setattr(py4j.java_gateway.JavaPackage, '__call__', wrapped_call)