Hello everyone i am pretty new in Apache Flink. I deployed flink-k8s-operator via helm inside of my eks cluster and i build custom pyflink image. My example applications are running without issue. But i need to consume data from RabbitMQ. While trying to connect RabbitMQ i am getting following error. My Pyflink code is;

import os
from pyflink.common import SimpleStringSchema
from pyflink.datastream.connectors.rabbitmq import RMQConnectionConfig, RMQSource
from pyflink.datastream.stream_execution_environment import StreamExecutionEnvironment

# logs configs
RABBITMQ_LOGS_HOST = os.getenv("RABBITMQ_LOGS_HOST")
RABBITMQ_LOGS_USER = os.getenv("RABBITMQ_LOGS_USER")
RABBITMQ_LOGS_PASSWD = os.getenv("RABBITMQ_LOGS_PASSWD")
RABBITMQ_LOGS_PORT = os.getenv("RABBITMQ_LOGS_PORT")
QUEUE_NAME = os.getenv("QUEUE_NAME")


env = StreamExecutionEnvironment.get_execution_environment()
env.add_jars("file:///opt/flink/lib/flink-connector-rabbitmq-1.16.1.jar","file:///opt/flink/lib/flink-sql-connector-rabbitmq-1.16.0")
env.enable_checkpointing(600000)

rabbitmq_connection_config = RMQConnectionConfig.Builder() \
    .set_host(RABBITMQ_LOGS_HOST) \
    .set_port(RABBITMQ_LOGS_PORT) \
    .set_user_name(RABBITMQ_LOGS_USER) \
    .set_password(RABBITMQ_LOGS_PORT) \
    .set_prefetch_count(30000).build()

My deployment yaml is;

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: wuxi
  namespace: flink
spec:
  ingress:
    template:  "my-private-ingress-url"
    className: "nginx"
    annotations: {}
  image: "my-private-repo"
  imagePullPolicy: "Always"
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "1"
  serviceAccount: flink
  jobManager:
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
       name: pod-template
      spec:
        serviceAccount: flink
        containers:
        # Do not change the main container name
          - name: flink-main-container
            envFrom:
            - secretRef:
                name: flink-secret
    resource:
      memory: "4096m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 1
    replicas: 1
  job:
    jarURI: local:///opt/flink/opt/flink-python_2.12-1.16.1.jar # Note, this jarURI is actually a placeholder
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-pyclientexec", "/usr/local/bin/python3", "-py", "/opt/flink/usrlib/main.py"]
    parallelism: 1
    upgradeMode: stateless

I am getting following errors; ERRORrootException while sending command. Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1224, in send_command raise Py4JNetworkError(“Answer from Java side is empty”) py4j.protocol.Py4JNetworkError: Answer from Java side is empty During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1038, in send_command response = connection.send_command(command) File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1228, in send_command raise Py4JNetworkError( py4j.protocol.Py4JNetworkError: Error while receiving Traceback (most recent call last): File “/usr/local/lib/python3.8/runpy.py”, line 194, in _run_module_as_main return _run_code(code, main_globals, None, File “/usr/local/lib/python3.8/runpy.py”, line 87, in _run_code exec(code, run_globals) File “/tmp/pyflink/cb21431412312412321412231/421312125676768445654642/main.py”, line 22, in <module> rabbitmq_connection_config = RMQConnectionConfig.Builder() \ File “/opt/flink/opt/python/pyflink.zip/pyflink/datastream/connectors/rabbitmq.py”, line 83, in init File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1547, in getattr py4j.protocol.Py4JError: org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig.Builder does not exist in the JVM ERROR org.apache.flink.runtime.util.ClusterUncaughtExceptionHandler [] - WARNING: Thread ‘Thread-14’ produced an uncaught exception. If you want to fail on uncaught exceptions, then configure cluster.uncaught-exception-handling accordingly java.lang.NoClassDefFoundError: com/rabbitmq/client/ConnectionFactory at java.lang.Class.getDeclaredMethods0(Native Method) ~[?:?] at java.lang.Class.privateGetDeclaredMethods(Unknown Source) ~[?:?] at java.lang.Class.privateGetPublicMethods(Unknown Source) ~[?:?] at java.lang.Class.getMethods(Unknown Source) ~[?:?] at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:284) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.getMember(ReflectionCommand.java:140) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.execute(ReflectionCommand.java:91) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) ~[flink-python-1.16.1.jar:1.16.1] at java.lang.Thread.run(Unknown Source) [?:?] Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.ConnectionFactory at jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source) ~[?:?] at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source) ~[?:?] at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?] ... 9 more 2023-03-04 070448,953 INFO org.apache.flink.client.python.PythonDriver [] - ERRORrootException while sending command. Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1224, in send_command raise Py4JNetworkError(“Answer from Java side is empty”) py4j.protocol.Py4JNetworkError: Answer from Java side is empty Can anyone help to solve this issue please ? Very thanks for your effort 🙂

Seems like rabbitmq is not on your Flink classpath. How do you build the dockerimage? Where did you put the rabbitmq connector jars?

Hello @Gyula Fóra, I built custom pyflink image and stored on private repository. After that, using this image i am creating new docker image with additional jars and python files. When i look under of the lib/ directory. I am able to see rabbitmq connector jars. Here is my Dockerfile;

FROM <private_repo>/pyflink:1.16.0

COPY jars/.       /opt/flink/lib/
ADD main.py    /opt/flink/usrlib/main.py

@Mali Hey, the

.jar

suffix is missing. It should be

file:///opt/flink/lib/flink-sql-connector-rabbitmq-1.16.0.jar

instead of

file:///opt/flink/lib/flink-sql-connector-rabbitmq-1.16.0

Hey @Dian Fu, I tried but i am getting same error 😕

@Mali Have you removed

file:///opt/flink/lib/flink-connector-rabbitmq-1.16.1.jar

?

yes i removed it

Could you post the latest exception stack?

ERRORrootException while sending command. Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1224, in send_command raise Py4JNetworkError(“Answer from Java side is empty”) py4j.protocol.Py4JNetworkError: Answer from Java side is empty During handling of the above exception, another exception occurred: Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1038, in send_command response = connection.send_command(command) File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1228, in send_command raise Py4JNetworkError( py4j.protocol.Py4JNetworkError: Error while receiving 2023-03-06 051359,459 ERROR org.apache.flink.runtime.util.ClusterUncaughtExceptionHandler [] - WARNING: Thread ‘Thread-14’ produced an uncaught exception. If you want to fail on uncaught exceptions, then configure cluster.uncaught-exception-handling accordingly java.lang.NoClassDefFoundError: com/rabbitmq/client/ConnectionFactory at java.lang.Class.getDeclaredMethods0(Native Method) ~[?:?] at java.lang.Class.privateGetDeclaredMethods(Unknown Source) ~[?:?] at java.lang.Class.privateGetPublicMethods(Unknown Source) ~[?:?] at java.lang.Class.getMethods(Unknown Source) ~[?:?] at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:284) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.getMember(ReflectionCommand.java:140) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.execute(ReflectionCommand.java:91) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) ~[flink-python-1.16.1.jar:1.16.1] at java.lang.Thread.run(Unknown Source) [?:?] Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.ConnectionFactory at jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source) ~[?:?] at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source) ~[?:?] at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?] ... 9 more Traceback (most recent call last): File “/usr/local/lib/python3.8/runpy.py”, line 194, in _run_module_as_main return _run_code(code, main_globals, None, File “/usr/local/lib/python3.8/runpy.py”, line 87, in _run_code exec(code, run_globals) File “/tmp/pyflink/adsadadsaafsad/12414123112314123/main.py”, line 18, in <module> rabbitmq_connection_config = RMQConnectionConfig.Builder() \ File “/opt/flink/opt/python/pyflink.zip/pyflink/datastream/connectors/rabbitmq.py”, line 83, in init File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1547, in getattr py4j.protocol.Py4JError: org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig.Builder does not exist in the JVM 2023-03-06 051359,542 INFO org.apache.flink.client.python.PythonDriver [] - ERRORrootException while sending command. Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1224, in send_command raise Py4JNetworkError(“Answer from Java side is empty”) py4j.protocol.Py4JNetworkError: Answer from Java side is empty

Are there still error messages such as

Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.ConnectionFactory

like before?

I changed “jarURI: local:///opt/flink/opt/flink-python_2.12-1.16.1.jar” to “jarURI: file:///opt/flink/lib/flink-sql-connector-rabbitmq-1.16.0.jar” in my deployment.yaml i think i passed the first exceptions. After all i am getting this error; ERROR org.apache.flink.runtime.util.ClusterUncaughtExceptionHandler [] - WARNING: Thread ‘Thread-14’ produced an uncaught exception. If you want to fail on uncaught exceptions, then configure cluster.uncaught-exception-handling accordingly java.lang.NoClassDefFoundError: com/rabbitmq/client/ConnectionFactory at java.lang.Class.getDeclaredMethods0(Native Method) ~[?:?] at java.lang.Class.privateGetDeclaredMethods(Unknown Source) ~[?:?] at java.lang.Class.privateGetPublicMethods(Unknown Source) ~[?:?] at java.lang.Class.getMethods(Unknown Source) ~[?:?] at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:284) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.getMember(ReflectionCommand.java:140) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.commands.ReflectionCommand.execute(ReflectionCommand.java:91) ~[flink-python-1.16.1.jar:1.16.1] at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) ~[flink-python-1.16.1.jar:1.16.1] at java.lang.Thread.run(Unknown Source) [?:?] Caused by: java.lang.ClassNotFoundException: com.rabbitmq.client.ConnectionFactory at jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source) ~[?:?] at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source) ~[?:?] at java.lang.ClassLoader.loadClass(Unknown Source) ~[?:?] ... 9 more ERRORrootException while sending command. Traceback (most recent call last): File “/opt/flink/opt/python/py4j-0.10.9.3-src.zip/py4j/java_gateway.py”, line 1224, in send_command raise Py4JNetworkError(“Answer from Java side is empty”) py4j.protocol.Py4JNetworkError: Answer from Java side is empty

I suspect that

flink-connector-rabbitmq-1.16.1.jar

still exists somewhere. In

flink-sql-connector-rabbitmq-1.16.1.jar

, the dependencies have been shaded and so the package name of

ConnectionFactory

should be

<http://org.apache.flink.rabbitmq.shaded.com|org.apache.flink.rabbitmq.shaded.com>.rabbitmq.client.ConnectionFactory

instead of ``com.rabbitmq.client.ConnectionFactory` . From the exception stack, it’s looking for

com.rabbitmq.client.ConnectionFactory

. This can occur when

flink-connector-rabbitmq-1.16.1.jar

also exsits in the classpath. Could you check the logging file where it will print the whole classpaths?

Yes you are right ! I deleted “flink-connector-rabbitmq-1.16.1.jar” as you said and the error is not exist. Currently, i am getting different error but let me try to solve it 🙂 I need to learn, very thanks @Dian Fu 🙏

I’m sorry that I’m not familiar with rabbitmq. From the exception message,

ACCESS_REFUSED

, it seems like permission issues.