Open indraneeldey1 opened 3 years ago
mhowlett
commented
3 years ago how did you make TestAvro.txt? It think it's because there's no _SCHEMA field, which there would be if AvroGen made it.
btw i would recommend protobuf over avro now we've added it - the implementation in .net is much more mature.
indraneeldey1
commented
3 years ago I made the original class via hand, so I redid the TestAvro.txt (really a cs file) using AvroGen and added that back in, though I get the same behavior. I also added a KeyDeserializer as well:
var keyDeserializer = new AvroDeserializer<string>(schemaRegistry).AsSyncOverAsync();
sing (var consumer = new ConsumerBuilder<string, TestAvro>(consumerConfig)
.SetKeyDeserializer(keyDeserializer)
.SetValueDeserializer(valueDeserializer)
.SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
.Build())Though now I get the same issue, just on the Key, its saying: Local: Key deserialization error instead. I tried to pause on the class methods through and found it would not go into the class.
Confluent.Kafka.ConsumeException: Local: Key deserialization error
---> System.NullReferenceException: Object reference not set to an instance of an object.
at Confluent.SchemaRegistry.Serdes.SpecificDeserializerImpl`1.Deserialize(String topic, Byte[] array)
at Confluent.SchemaRegistry.Serdes.AvroDeserializer`1.DeserializeAsync(ReadOnlyMemory`1 data, Boolean isNull, SerializationContext context)
at Confluent.Kafka.SyncOverAsync.SyncOverAsyncDeserializer`1.Deserialize(ReadOnlySpan`1 data, Boolean isNull, SerializationContext context)
at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)
--- End of inner exception stack trace ---
at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)
at Confluent.Kafka.Consumer`2.Consume(CancellationToken cancellationToken)
at Kafka.Consumer.Program.Main(String[] args) in C:\Users\idey\RiderProjects\Kafaka\Kafka.Consumer\Program.cs:line 51Also just to confirm I was able to hit http://localhost:8081/schemas/ids/1 via PostMan and was returned the following:
{
"schema": "{\"type\":\"record\",\"name\":\"test\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"},{\"name\":\"name\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"email\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"department\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"modified\",\"type\":{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}}],\"connect.name\":\"test\"}"
}I'll look into swapping over Protobuf as well.
Description
Hi there, I've been noticing that I've been getting a Null Exception error from the AvroDeserializer that I have not been able to resolve, though data comes though. I'm following the AvroSecific example found at https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples/AvroSpecific
It followed the base tutorial of working with Docker: https://docs.confluent.io/5.0.0/installation/docker/docs/installation/connect-avro-jdbc.html
Though when consuming as a Generic record, no problems are found. I have found similar issues, but not one like this specifically from what errors I've been able to find.
How to reproduce
Follow example of AvroSpecific as stated before, using the AvroDeserializer
Checklist
Please provide the following information:
Basics
Found Errors
The initial error I was able to find that I caught inside for my own exception was the following:
Digging further inside of the Consumer Builder -> Consume Function on line 682 an exception is given:
The inspection of this.valueDeserializer does show that deserializeImpl is nulled.
Docker Compose Configuration
docker-compose.txt
I've attached the Docker Compose file, that contains Zookeeper, Kafka, Schema Registry, Kafka Connect, MySQL.
Consumer program
Program.txt
The consume program follows the AvroSpecific example rather closely, though did split out the Deserializers as their own variables.
Avro Model
TestAvro.txt
The Test Avro file follows the example rather closely, the schema was grabbed through the SchemaRegistry API, though I did add in the namespace.
Data
Starting a bin session into the SchemaRegistry container I ran the following command:
kafka-avro-console-consumer --bootstrap-server kafka:9092 --topic quickstart-jdbc-test --property schema.registry.url="http://schema-registry:8081" --from-beginning --max-messages 10and through that I got the following:
Big thanks for reading through this!