Issue While consuming the messages from an Avro Topic

[AmbiliVaradabai]

Hi All,
I am facing an issue while consuming the data from an avro topic.
Can anyone provide an example for the same. I am able to consume without giving the schema regitry url, but the data is not in readable format. When the schema regitry url is given I am not able to consume messages.

[robooo]

Hi Ambili,
Right now the logic is, when you don't set the schema_registry_url (None by default) the Consumer is used instead of AvroConsumer. You need to have this schema to decode data (check the confluent kafka docu) and the url to this schema should be accessible from where are you running this library.

Could you be more specific about that you are not able to consume messages? like are you getting some errors? are you sure the data are still streaming to the topic? (Try set the auto_offset_reset='earliest' if you want to read older data), and provide youre example which doesn't work.

Personally I'm using it in my test suite like:

Create Avro Consumer In Thread
    ${thread}=  Start Consumer Threaded  topics=my_topic_where_data_are_streaming
        ...  schema_registry_url=http://schemaregistry.${DOMAIN}
        ...  auto_offset_reset=earliest

Or secure version

Create Secure Avro Consumer In Thread
    ${thread}=  Start Consumer Threaded  topics=my_topic
        ...  schema_registry_url=https://schemaregistry.${DOMAIN}
        ...  auto_offset_reset=earliest
        ...  security.protocol=ssl
        ...  schema.registry.ssl.ca.location=${CA_PROXY_CERT_FILE}
        ...  ssl.ca.location=${CA_PROXY_CERT_FILE}
        ...  ssl.certificate.location=${KAFKA_CLIENT_CERT_FILE}
        ...  ssl.key.location=${KAFKA_CLIENT_CERT_KEY_FILE}

and then when I want to check during testing if I have a record in that topic I use:

${messages}=  Get Messages From Thread  ${thread}
    FOR  ${message}  IN  @{messages}
      Run Keyword If  '${device_id}' == '${message['device_id']}' and '${status.lower()}' == '${message['status'].lower()}'
    END
...

Edit: I have many new updates + changes in the other branch, right now I will have more time so there will be more examples too ;)

[AmbiliVaradabai]

Hi Robert,
Thank you for the quick response.
I am not getting any error messages, When I tried to log the message to console I am not seeing anything. In fact I followed the above same steps using thread as well.
I tried something like below

${thread}= Start Consumer Threaded topics=test-topic
schema_registry_url=http://hostIP:port
schema.registry.ssl.ca.location=%{LOCATION}
auto_offset_reset=earliest
security.protocol=%{PROTOCOL}
sasl.mechanism=%{MECHANISM}
sasl.username=%{USERNAME}
sasl.password=%{PASSWORD}
ssl.ca.location=%{LOCATION}
${messages}= Get Messages From Thread ${thread}
Log To Console ${messages}
Stop Thread ${thread}

However, after execution came to log. it was just "[]"

My intention is to get one message from an avro topic and compare it against another topic message, I am able to do this successfully for the topics with text messages and json messages.

I also tried to consume the avro topic in the same was as the text topics are consumed with out schema registry url and it was giving me an output in the Log. However that was not readable as it contains strings like "X00/"

[robooo]

that \x0 should be ok those are null bytes, and inside of that string you should see other bytes as well, and that should be the coded data you are looking for.
Could you verify that schema_registry_url=http://hostIP:port does opens in your web browser and returns schema in json-like format? if not try the https or check if you don't have the schema at specific url like https://schemaregistry.hostIP:port

[AmbiliVaradabai]

When \X0 is there it is difficult to read the message ans compare.
In order to view the message in avro format, along with schema registry url do we need to pass any decoding mechanism?

[robooo]

you can try decode_format='utf-8', remove_zero_bytes=True
https://github.com/robooo/robotframework-ConfluentKafkaLibrary/blob/master/src/ConfluentKafkaLibrary/consumer.py#L230

[AmbiliVaradabai]

I am trying below consumer code now to consume avro data

${group_id}= Create Consumer server=${Server} port=${Port} auto_offset_reset=earliest security.protocol=SASL_SSL sasl.mechanism=%{mechanism} sasl.username=%{USERNAME} sasl.password=%{PASSWORD} ssl.ca.location=%{LOCATION}
Subscribe Topic group_id=${group_id} topics=AvroTopic
${ActualOutputmessages}= Poll group_id=${group_id} max_records=1
Set Global Variable ${ActualOutputmessages}
#${messages}= Decode Data ${ActualOutputmessages} decode_format=utf8
Log To Console ${ActualOutputmessages}
Unsubscribe ${group_id}
Close Consumer ${group_id}

When I am running above code, as I mentioned earlier I am getting a string out put with " \X0".
As suggested in the documention, I tried to add schema registry url but the log message was showing [] .
As shown in the above snippet, when I am uncommenting the decode data command, it is throwing error "'utf-8' codec can't decode byte 0xad in position 4: invalid start byte".

[AmbiliVaradabai]

Hello Robert,
Could you help me wioth one example of Avro producer.
What are the value_schema_str and key_schema_str ? Are this should be passed in JSON format?
Could you give me one example please?

[robooo]

@AmbiliVaradabai
I'm using that one I posted before, it runs with our production kafka like that during the testing via RF.

• Have you verified if the schema is visible at that endpoint you are setting in schema_registry_url ?
• To fix that utf-8 err, you can set remove_zero_bytes=True as I mentioned before (but this will be removed in future release because it doesn't make sense in this library, was created only for debug purpose)
• Try to execute poll twice (copy the same poll line after), maybe you have empty message in kafka? or are you sure there are messages?

I'm sorry that value_schema_str and key_schema_str were removed (but not from comment...) you can ignore that part.

You can check the official documentation here, https://docs.confluent.io/current/clients/confluent-kafka-python/#module-confluent_kafka.avro
and read about all configuration here https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

Try it with this python confluent kafka library and if it will work send me the code and I can look deeper, if it won't work with confluent kafka then the problem will be probably at your side (the configuration of kafka or some missing argument).

[AmbiliVaradabai]

Hi Robert,
Thank you for your suggestion.

In order to produce data, I have followed the official documentation as you mentioned above and was able to successfully produce avro data to a topic.
Used the below example.
https://github.com/skyrocknroll/python-kafka-avro-example/blob/master/avro-producer.py

However, could you please help me to understand how to achieve this using the ConfluentKafkaLibrary, because if it is successful, I will be be able to do this in a single step.

When tried to use the Library, I passed the value and key schema as below

${producer_group_id}= Create Producer server=${Server} port=${Port} schema_registry_url=http://domain:port value_schema_str=${valueSchema} key_schema_str=${KeySchema} security.protocol=${protocol} sasl.mechanism=${mechanism} sasl.username=${UserName} sasl.password=${password} ssl.ca.location=${CA_Location}

I am getting below error using the schema from the example "https://github.com/skyrocknroll/python-kafka-avro-example/blob/master/avro-producer.py"

ClientError: Schema parse failed: Error parsing schema from JSON: {'type': 'record', 'name': 'User', 'namespace': 'example.avro', 'fields': [{'type': 'string', 'name': 'name'}, {'type': ['int', 'null'], 'name': 'favorite_number'}, {'type': 'string', 'name': 'favorite_color'}, {'type': 'int', 'name': 'age', 'default': 18}]}. Error message: TypeError("the JSON object must be str, bytes or bytearray, not 'dict'",).

Could you help me here please.
Kindly let me know if any additional details are required.

[robooo]

Hi again :) please the issue was related to consumer now it's producer and it looks like you have wrong input but maybe it will contains some issue ;)

ok so you have nice description to error: you need to convert json to str, so you can use str() function but probably you have wrong data in argments, Try to Log ${valueSchema} and Log ${KeySchema} what's inside (you can post it here), but it should contains only path to asc file as here, not already loaded data:
https://github.com/skyrocknroll/python-kafka-avro-example/blob/master/avro-producer.py#L6

[AmbiliVaradabai]

Thanks again :)
There are two issues that I am facing
Issue #1 :
Not able to consume the avro topic data with schema registry url. I have a work work around for that. I have used normal consumer ( without schema registry url) and then wrote a custom library to decode data in to avro format.

Issue # 2 : This is more important for me now .I am unable to produce avro data using the library.
The string error mentioned in my previous comment is resolved and now I am getting another error. The code is something like below

*** Settings ***
Library ConfluentKafkaLibrary
Library Collections
Library OperatingSystem

*** Variables ***
${Key} {"name": "Robot"}
${Value} {"name": "Robot", "favorite_number": 10, "favorite_color": "green", "age": 25}

*** Test Cases ****
Producing avro data to topic

${Value_Schema}= Get File ValueSchema.avsc
${Key_Schema}= Get File KeySchema.avsc

${producer_group_id}= Create Producer server=##.###.##.# port=####
schema_registry_url=http://##.##.#.#:#### value_schema_str=${Value_Schema}
key_schema_str=${Key_Schema} security.protocol=SASL_SSL sasl.mechanism=#####
sasl.username=##### sasl.password=### ssl.ca.location=####
Produce group_id=${producer_group_id} topic=Avro-Test-Topic value=${Value} key=${Key}
Flush ${producer_group_id}

My avsc files are just same as the one in the below example
https://github.com/skyrocknroll/python-kafka-avro-example/tree/master/schema/producer

When I am running the robot script, I am getting below error

AvroProducer

Producing avro data to topic | FAIL |
TypeError: a bytes-like object is required, not 'str'

AvroProducer | FAIL |
1 critical test, 0 passed, 1 failed
1 test total, 0 passed, 1 failed

I know I am missing something. Could you please guide me what is missing here.
Thanks again for all all the inputs.

[robooo]

and please you still haven't answered to #1:
is that schema visible in the web browser? and does it works with https://github.com/skyrocknroll/python-kafka-avro-example/blob/master/avro-consumer.py ?

#2:
there could be some catch https://github.com/robooo/robotframework-ConfluentKafkaLibrary/blob/master/src/ConfluentKafkaLibrary/producer.py#L94
it looks like it needs object so I don't get the error, could you paste here the full error from RF log? because I think this is happening for key (will need that encoding to auto transfer to bytes probably)

and try something like I did here https://github.com/robooo/robotframework-ConfluentKafkaLibrary/blob/updates/examples/test.robot#L126
so

Produce group_id=${producer_group_id} topic=Avro-Test-Topic value={"name": "Robot", "favorite_number": 10, "favorite_color": "green", "age": 25} key={"name": "Robot"}

but you will need transfer the {"name": "Robot"} to bytes if I'm right so you will need to transfer it to bytes

Produce group_id=${producer_group_id} topic=Avro-Test-Topic value={"name": "Robot", "favorite_number": 10, "favorite_color": "green", "age": 25} key=b'{"name": "Robot"}'

if this is true I will create issue for this ;) and will fix in next release

[AmbiliVaradabai]

Hi Robert,
Please find below details .
Issue # 1 : The schema is visible in the web browser.
Tried with the suggested code and looks like it is able to consume the data but the serialization is not working. Getting below error
"Message deserialization failed for None: Message deserialization failed for message at Topic [0] offset 5: message does not start with magic byte"

Issue # 2 : Tried with the above step using "Produce group_id=${producer_group_id} topic=Avro-Test-Topic value={"name": "Robot", "favorite_number": 10, "favorite_color": "green", "age": 25} key=b'{"name": "Robot"}' ", still I am getting the same error.

Below is the full error

AvroProducer

Producing avro data to topic | FAIL |
TypeError: a bytes-like object is required, not 'str'

AvroProducer | FAIL |
1 critical test, 0 passed, 1 failed
1 test total, 0 passed, 1 failed

[robooo]

Paste the full log of

Producing avro data to topic | FAIL |
TypeError: a bytes-like object is required, not 'str'

from the RF log or screenshot, as you can see it's not so verbose :)

Edit:
sorry to #1

Message deserialization failed for None

so you have empty message, which can't be deserialized, that means you get same with this RF wrapper too.

[AmbiliVaradabai]

Please find attached screen shot