Java types and mapping of avro types

[cagalmarini]

Hi, I'm working on some proof of concept involving Apicurio, Avro and Kafka, using Java
Lets suppose I have a POJO which contains a Date field that are being sent to Kafka as Avro.
When those objects get automatically registered on apicurio because of apicurio.registry.auto-register being true those schemas are effectively created but the Date member is mapped as java.util.Date instead of "type": {"type": "int", "logicalType": "date"}.
POJO:

package avrobeanexample;

public class MyTestBean {
	private String message;
	private java.time.LocalDate date;
	
	public MyTestBean() {
		
	}
	
	public String getMessage() {
		return message;
	}
	
	public void setMessage(String message) {
		this.message = message;
	}
	
	public java.time.LocalDate getDate() {
		return date;
	}
	
	public void setDate(java.time.LocalDate date) {
		this.date = date;
	}

}

Registered schema:

{
  "type": "record",
  "name": "MyTestBean",
  "namespace": "avrobeanexample",
  "fields": [
    {
      "name": "date",
      "type": "java.time.LocalDate"
    },
    {
      "name": "message",
      "type": "string"
    }
  ]
}

Kafka producer configuration:

private static Producer<Object, Object> createKafkaProducer() {						
	Properties props = new Properties();

        props.putIfAbsent(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, SERVERS);
        props.putIfAbsent(ProducerConfig.CLIENT_ID_CONFIG, "Producer-" + TOPIC_NAME);
        props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "all");
        props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); 	
        props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AvroKafkaSerializer.class.getName());

        props.putIfAbsent(SerdeConfig.REGISTRY_URL, REGISTRY_URL);							
        props.putIfAbsent(SerdeConfig.AUTO_REGISTER_ARTIFACT, Boolean.TRUE);										

        props.putIfAbsent(AvroKafkaSerdeConfig.AVRO_DATUM_PROVIDER, "io.apicurio.registry.serde.avro.ReflectAvroDatumProvider");

        Producer<Object, Object> producer = new KafkaProducer<>(props);							

        return producer;
    }

Records are succesfully produced against Kafka but consuming them throws an exception:

Exception in thread "main" org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition avrobeanexample-2 at offset 15. If needed, please seek past the record to continue consumption.
	at org.apache.kafka.clients.consumer.internals.CompletedFetch.parseRecord(CompletedFetch.java:309)
	at org.apache.kafka.clients.consumer.internals.CompletedFetch.fetchRecords(CompletedFetch.java:263)
	at org.apache.kafka.clients.consumer.internals.AbstractFetch.fetchRecords(AbstractFetch.java:340)
	at org.apache.kafka.clients.consumer.internals.AbstractFetch.collectFetch(AbstractFetch.java:306)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1262)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1186)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1159)
	at avrobeanexample.AvroBeanExample.main(AvroBeanExample.java:64)
Caused by: java.lang.RuntimeException: java.lang.RuntimeException: java.lang.NoSuchMethodException: java.time.LocalDate.<init>()
	at org.apache.avro.specific.SpecificData.newInstance(SpecificData.java:493)
	at org.apache.avro.specific.SpecificData.newRecord(SpecificData.java:511)
	at org.apache.avro.reflect.ReflectData.newRecord(ReflectData.java:1057)
	at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237)
	at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
	at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
	at org.apache.avro.reflect.ReflectDatumReader.readField(ReflectDatumReader.java:301)
	at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:248)
	at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
	at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
	at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:154)
	at io.apicurio.registry.serde.avro.AvroKafkaDeserializer.readData(AvroKafkaDeserializer.java:117)
	at io.apicurio.registry.serde.AbstractKafkaDeserializer.readData(AbstractKafkaDeserializer.java:142)
	at io.apicurio.registry.serde.AbstractKafkaDeserializer.deserialize(AbstractKafkaDeserializer.java:122)
	at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:73)
	at org.apache.kafka.clients.consumer.internals.CompletedFetch.parseRecord(CompletedFetch.java:300)
	... 7 more

By the way, if I use the avro maven plugin to generate Java classes based on a schema file on disk that class is created using LocalDate as member type for schema fields that are logical type date.
Example schema on disk:

{
    "type": "record",
    "name": "AuthorDemo",
    "namespace": "mycompany",
    "fields": [
        {
            "name": "dob",
            "type": {"type": "int", "logicalType": "date"}
        },
        {
            "name": "name",
            "type": "string"
        }
    ]
}

Class piece of code generated:

...
...
...
private java.time.LocalDate dob;
  private java.lang.String name;

  /**
   * Default constructor.  Note that this does not initialize fields
   * to their default values from the schema.  If that is desired then
   * one should use <code>newBuilder()</code>.
   */
  public AuthorDemo() {}

  /**
   * All-args constructor.
   * @param dob The new value for dob
   * @param name The new value for name
   */
  public AuthorDemo(java.time.LocalDate dob, java.lang.String name) {
    this.dob = dob;
    this.name = name;
  }
...
...
...

Any ideas about how to solve this issue?
Thanks