Hi, right now I am trying to run a pyflink script that can connect to a kafka server. When I run that script, I got an error "An error occurred while trying to connect to the Java server 127.0.0.1:35529". Do I need to install a extra jdk for that?

error message:

Py4JNetworkError: An error occurred while trying to connect to the Java server (127.0.0.1:35529)

---------------------------------------------------------------------------

IndexError Traceback (most recent call last)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in _get_connection(self)

976 try:

--> 977 connection = self.deque.pop()

978 except IndexError:

IndexError: pop from an empty deque

During handling of the above exception, another exception occurred:

ConnectionRefusedError Traceback (most recent call last)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in start(self)

1114 try:

-> 1115 self.socket.connect((self.address, self.port))

1116 self.stream = self.socket.makefile("rb")

ConnectionRefusedError: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:

Py4JNetworkError Traceback (most recent call last)

<command-1981093759856396> in <module>

88

89

---> 90 main()

<command-1981093759856396> in main()

5 def main():

6 # Create streaming environment

----> 7 env = StreamExecutionEnvironment.get_execution_environment()

8

9 settings = EnvironmentSettings.new_instance()\

/databricks/python/lib/python3.8/site-packages/pyflink/datastream/stream_execution_environment.py in get_execution_environment()

801 """

802 gateway = get_gateway()

--> 803 j_stream_exection_environment = gateway.jvm.org.apache.flink.streaming.api.environment\

804 .StreamExecutionEnvironment.getExecutionEnvironment()

805 return StreamExecutionEnvironment(j_stream_exection_environment)

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in __getattr__(self, name)

1690 return UserHelpAutoCompletion()

1691

-> 1692 answer = self._gateway_client.send_command(

1693 proto.REFLECTION_COMMAND_NAME +

1694 proto.REFL_GET_UNKNOWN_SUB_COMMAND_NAME + name + "\n" + self._id +

/databricks/spark/python/lib/py4j-0.10.9.1-src.zip/py4j/java_gateway.py in send_command(self, command, retry, binary)

1029 if `binary` is `True`.

1030 """

-> 1031 connection = self._get_connection()

1032 try: