[Bug]: ReadFromKafka not forwarding in streaming mode version on portable runners

[jihad-akl]

What happened?

ReadFromKafka not forwarding in streaming mode.
using apache-beam 2.44.0

beam_options = PipelineOptions(streaming = True)
pipeline = beam.Pipeline(options=beam_options)

messages = (
pipeline
| 'Read from Kafka' >> ReadFromKafka(
consumer_config=json.load(open("config/consumer_config_beam.json")),
topics=topic
)
| 'Print messages' >> beam.Map(lambda message: print("received!"))
)

Hello, in the code above, the code is stuck on ReadFromKafka.
Adding max_num_records will only wait for the specific amount of data and them forward them to the next step and ends the codes.
(I am using the DirectRunner I need to run the code locally)

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

streaming by definition will not end; despite python directly runner is not for production and do not have full support for streaming. This is most likely working as intended

[jihad-akl]

True, so how can use the apache beam pipeline in streaming mode if it only gather data and not send them to the next step?
The print received does not trigger every time I receive a message from kafka locally. Isn't it an important bug? and the direct runner need some fixes?

[tvalentyn]

#24528 tracks various issues related to streaming direct runner. I am not sure if it is able to run a simple KafkaIO pipeline. Are you able to use a portable Flink Runner by chance

[jihad-akl]

I am trying to to implement it but till now I am facing the same issue, I can see my pipeline in the apache flink localhost:8081 but nothing happens, I am debugging it to see if I made any mistake

[jihad-akl]

So after researching and testing, I found that the Flink Runner does not help because apache flink gather a lot of data before releasing them, my use case is every message I receive from kafka I need to forwarded to the next step in the pipeline (locally).

[Abacn]

fyi if flink runner has the same issue, it may hit #22809, the issue in python side may still persist
CC: @johnjcasey
I will also take a look