How to consume messages using Kafkastrem

Hi all,

The question is simple as the title says. How to consume messages from selected Kafka topic?

I have a setup which works perfectly from terminal and now I am trying to connect MATLAB with this Kafka topic, just to consume messages. I followed this:

https://www.mathworks.com/help/mps/ml_code/matlab.io.stream.event.kafkastream.html

but unfortunately there is no information how to consume messages. Any help will be appreciated!

[Walter Roberson]

What if you seek() to end ? https://www.mathworks.com/help/mps/ml_code/matlab.io.stream.event.kafkastream.seek.html

Or https://www.mathworks.com/help/mps/ml_code/matlab.io.stream.event.kafkastream.flush.html flush() ?

[Marcin Kowalski]

I believe the issue is somewhere else. I have two machines - one with Kafka server and the second with Kafka consumer (WSL2). Communication between both works perfectly, but I cannot do the same using MATLAB.

I can set the KafkaStream object - that works well

host="10.1.147.182";
port=9092;
topic="demo-messages";

ks = kafkaStream(host,port,topic, Rows=10)

but then, I try to consume messages by

event = readevents(ks)

preview(ks)

What happens is that MATLAB is busy for ~1 min and replies with 0x0 empty timetable. Any ideas what I am doing wrong here?

Answer 1

[Divyanshu]

Hi Marcin,

I understand that you are able to configure the kafkaStream object successfully but are facing issues while reading the messages from KafkaStream.

Please go through the following documentation which provides an example of writing and reading messages over the kafka stream:

https://www.mathworks.com/help/mps/ml_code/process-kafka-events-using-matlab.html

Also make sure to verify that once the kafkaStream object is configured, the messages are written appropriately. The function 'readtimetable' can be used to read the data as shown in the example above.

For further details about 'readtimetable' function refer the following documentation:

https://www.mathworks.com/help/mps/ml_code/matlab.io.stream.event.kafkastream.readtimetable.html

[Marcin Kowalski]

Thank you for the answer.

In fact there was a misconfiguration of the kafka server which has been solved and I can read the messages correctly.

But another challenge appeared which is the limitation of the number of messages possible to read. I managed to successfully read 50 messages form the selected topic, but not more. Teoretically it is possible to change the number of rows. but when I increase the number over 50, the reading time is very long, couple of minutes. For over 100 I have not waited, it took that long.

ks = kafkaStream(host,port,topic,Rows=numevents)

Any ideas?

[Divyanshu]

Hi Marcin,

This particular issue requires further investigation and would require some log files to track the root cause of the issue and provide a solution or workaround accordingly.

Hence I would suggest you to reach out to MathWorks Technical Support and raise a request on the below link:

https://in.mathworks.com/support/contact_us.html

Once a request is raised someone from technical support would reach out to you and would help you with the issue.

Thanks.

[Marcin Kowalski]

Thank you, request sent.