A simple code to read data from my local Kafka cluster using DataStream API. Unable to figure out why the execution environment is unable to read the data from Kafka and print the results. Thanks in advance for your help Flink fam !!!

from pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.common import WatermarkStrategy, SimpleStringSchema

def read_from_kafka(env):
    kafka_source = KafkaSource \
    .builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_topics("mqtt-replay") \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .set_group_id("my-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .build()

    env.add_source(kafka_source).print()
    env.execute()

if __name__ == '__main__':
    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_parallelism(1)
    env.add_jars("file:///Users/Raghav/Desktop/DaftPunk/Resources/flink-sql-connector-kafka-1.17.1.jar")
    print("Start reading data from kafka: ")
    read_from_kafka(env)

Getting error message - Traceback (most recent call last): File "flink.py", line 24, in <module> _read_from_kafka(env)_ _File "flink.py", line 17, in read_from_kafka_ env.execute() _File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/pyflink/datastream/stream_execution_environment.py", line 763, in execute_ _j_stream_graph = self._generate_stream_graph(clear_transformations=True, job_name=job_name)_ _File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/pyflink/datastream/stream_execution_environment.py", line 1012, in _generate_stream_graph_ _j_stream_graph = self._j_stream_execution_environment.getStreamGraph(clear_transformations)_

Have you pasted the entire error trace?

full error message- File "flink.py", line 30, in <module> read_from_kafka(env) File "flink.py", line 18, in read_from_kafka env.add_source(kafka_source).print() File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/pyflink/datastream/stream_execution_environment.py", line 882, in add_source j_data_stream = self._j_stream_execution_environment.addSource(source_func File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/py4j/java_gateway.py", line 1322, in call return_value = get_return_value( File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/pyflink/util/exceptions.py", line 146, in deco return f(*a, **kw) File "/Users/Raghav/Desktop/DaftPunk/Repos/Asset Tracking/asset-flink/flink/lib/python3.8/site-packages/py4j/protocol.py", line 330, in get_return_value raise Py4JError( py4j.protocol.Py4JError: An error occurred while calling o10.addSource. Trace: org.apache.flink.api.python.shaded.py4j.Py4JException: Method addSource([class org.apache.flink.connector.kafka.source.KafkaSource, class java.lang.String, null]) does not exist at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:321) at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:329) at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:274) at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) at java.base/java.lang.Thread.run(Thread.java:834)

I am also getting the same error irrespective of whether you include add_jars code or not.

I'm wondering if it's related to add_source or add_jars

if you remove add_jars, the error will remain the same since you have added the jar in the lib of pyflink. I suppose.

@Siddhesh Kalgaonkar will update once I try / resolve this

sure @dp api

Does not seem to be an issue with additional jars file either. I have flink v 1.17.1 python library as well as the same jar version. I think there is some issue with KafkaSource class. Do let me know if you are able to resolve it too.

I definitely think there is some version issues.

Versioning issue among apache-flink library in python and jar file ?

Could be anything. I highly doubt w.r.t Kafka client.

Thanks for sharing. Again no solution posted here. But I was able to read data from Kafka to Flink finally. It required using the FlinkKafkaConsumer even in v1.17.1 , even though the docs mentioned it being deprecated v1.16 onwards. Basically it seems like KafkaSource object may be a bit buggy.

Can you post the code here @dp api

Thanks I will check this.