Using Apache Hop integrated with Kafka

[eleonora011]

I’d like to briefly explain the workflow I developed in Apache Hop and the issues I’m encountering.

The flow works as follows:
It starts with a Kafka Consumer that reads messages from a specific topic. This task triggers a specific pipeline, which begins with an Injector. Then, to properly format each row, I use a JSON Input step and several Select Values steps.

My goal is to enrich the message coming from Kafka with data from the database. So, after a Table Input, I perform a Merge (between the data coming from Kafka and the data coming from the database), and finally, I output the result to a text file.

The problem I’ve encountered is that the flow stops at the Merge step.
The behavior is quite strange:
if I pass a finite number of rows to the pipeline, it completes successfully;
however, if I use a Generate Rows step (with rows generated at regular time intervals) or a Kafka Consumer (both of which can be considered “infinite” sources ) the flow freezes.

I’d like to know whether this issue has already been experienced by others, and if so, how it can be resolved or worked around.

Thank you

[hansva]

Could you add a reproduction case please? This would help us investigate the issue. I am guessing you are talking about a merge join? I would expect that one to pass the rows after the "group" is finished.

[eleonora011]

I have the main_kafka_pipeline, which only contains the Kafka Consumer, and it calls the following pipeline:

Injector -> JSON Input -> Select Values -> Sort Rows-----------------------v
----------------------------- ↓ --------------------------------------- Multiway Merge Join -> Text File Output
----------------------------- Select Values -> Table Input -> Sort Rows-----^

But since the Kafka Consumer is always listening, when would the “group” ever be considered finished?

[liosc]

I think you should prevent to use transform that wait for the end of the stream (like sort, merge, ...) in the same pipeline where the Kafka consumer transform is. Try to use a pipeline executor and put in the called pipeline the logic to manage the Kafka message.

[dmainou]

This looks like the right answer.

By using the pipeline executor you enable batching therefore your sort, at the batch level, works.

[eleonora011]

Thank you so much, it actually works in this way!

[dmainou]

BTW,
If you are looking into executing your pipeline as a streaming job (e.g. through Spark, Google Dataflow, Flink, etc)
Here is where you control your window either by:

a) number of rows
b) a column such as a session id, hour, etc
c) time