Search code examples
apache-kafkaksqldb

Error while consuming AVRO Kafka Topic from KSQL Stream

I created some dummydata as a Stream in KSQLDB with VALUE_FORMAT='JSON' TOPIC='MYTOPIC'

The Setup is over Docker-compose. I am running a Kafka Broker, Schema-registry, ksqldbcli, ksqldb-server, zookeeper

Now I want to consume these records from the topic. My first and last approach was over the commandline with following command

docker run  --net=host  --rm  confluentinc/cp-schema-registry:5.0.0  kafka-avro-console-consumer
--bootstrap-server localhost:29092 --topic DXT --from-beginning --max-messages 10
--property print.key=true --property print.value=true
--value-deserializer io.confluent.kafka.serializers.KafkaAvroDeserializer
--key-deserializer org.apache.kafka.common.serialization.StringDeserializer

But that just returns the error

[2021-04-22 21:45:42,926] ERROR Unknown error when running consumer:  (kafka.tools.ConsoleConsumer$:76)
org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

I also tried it with different use cases in Java Spring but with no prevail. I just cannot consume the created topics. If I would need to define my own schema, where should I do that and what would be the easiest way because I just created a stream in Ksqldb? Is there an easy to follow example. I did not specifiy anything else when I created the stream like in the quickstart example on Ksqldb.io. (I added the schema-registry in my deployment) As I am a noob that is sitting here for almost 10 hours any help would be appreciated.

Edit: I found that pure JSON does not need the Schema-registry with ksqldb. Here. But how to deserialize it?

Solution

  • If you've written JSON data to the topic then you can read it with the kafka-console-consumer.

    The error you're getting (Error deserializing Avro message for id -1…Unknown magic byte!) is because you're using the kafka-avro-console-consumer which attempts to deserialise the topic data as Avro - which it isn't, hence the error.

    You can also use PRINT DXT; from within ksqlDB.