CODEWITHSUNDEEP
apache-kafkaavroapache-kafka-streamsconfluent-schema-registry
Varun Bajaj
Varun Bajaj

Reputation: 1043

Deserializing exception with data generated ksql-datagen utility

Generated sample stream from ksql-datagen utility from following schema -

{
        "type": "record",
        "name": "users",
        **"namespace": "com.example",**
        "fields": [
        {
        "name": "registertime",
        "type": {
            "type":"long",
            "arg.properties":{
                "range":{"min":1487715775521,"max":1519273364600}
                }
        }
        },
        {
                "name": "userid",
                "type": {
            "type":"string",
            "arg.properties":{"regex":"User_[1-9][0-2]"}
        }
        },
        {
                "name": "regionid",
                "type": {
            "type":"string",
            "arg.properties":{"regex":"Region_[1-9]"}
        }
        },
        {
                "name": "gender",
                "type": {
            "type":"string",
            "arg.properties":{
            "options":["MALE","FEMALE","OTHER"]
            }
        }
        }
]}

while checked for versions, it still picks "io.confluent.ksql.avro_schemas" schema -

curl "http://localhost:8081/subjects/test-user-value/versions/1"

{"subject":"test-user-value","version":1,"id":4,"schema":"{"type":"record","name":"KsqlDataSourceSchema","namespace":"io.confluent.ksql.avro_schemas","fields":[{"name":"registertime","type":["null","long"],"default":null},{"name":"userid","type":["null","string"],"default":null},{"name":"regionid","type":["null","string"],"default":null},{"name":"gender","type":["null","string"],"default":null}]}"}

Got following error while tried to consume with Kafka-streams API -

Exception in thread "PageView-Users-Stream-Join-eg-1dc610a3-c9d9-4c1e-b5eb-910e4bc74826-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately. at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:80) at org.apache.kafka.streams.processor.internals.RecordQueue.maybeUpdateTimestamp(RecordQueue.java:160) at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:101) at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:124) at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:711) at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:995) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:833) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747) Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 4 Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class io.confluent.ksql.avro_schemas.KsqlDataSourceSchema specified in writer's schema whilst finding reader's schema for a SpecificRecord.

Upvotes: 0

Views: 669

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 191738

Answered at https://github.com/confluentinc/schema-registry/issues/980

Datagen always defines the namespace as io.confluent.ksql.avro_schemas. See confluentinc/ksql#1906

There are now other ways to generate test data into Kafka, too.

Upvotes: 1

Related Questions