[Avro] Basic logicalType support for date time types

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/AvroInstantDeserializer.java

@@ -0,0 +1,78 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.BeanProperty;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.deser.ContextualDeserializer;
+import com.fasterxml.jackson.databind.deser.std.StdScalarDeserializer;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.temporal.Temporal;
+import java.util.function.BiFunction;
+
+/**
+ * A deserializer for variants of java.time classes that represent a specific instant on the timeline
+ * (Instant, OffsetDateTime, ZonedDateTime) which supports de-serialization from Avro long.
+ *
+ * See: http://avro.apache.org/docs/current/spec.html#Logical+Types
+ *
+ * Note: {@link AvroInstantDeserializer} does not support deserialization from string.
+ *
+ * @param <T> The type of a instant class that can be deserialized.
+ */
+public class AvroInstantDeserializer<T extends Temporal> extends StdScalarDeserializer<T>
+        implements ContextualDeserializer {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final AvroInstantDeserializer<Instant> INSTANT =
+            new AvroInstantDeserializer<>(Instant.class, (instant, zoneID) -> instant);
+
+    public static final AvroInstantDeserializer<OffsetDateTime> OFFSET_DATE_TIME =
+            new AvroInstantDeserializer<>(OffsetDateTime.class, OffsetDateTime::ofInstant);
+
+    public static final AvroInstantDeserializer<ZonedDateTime> ZONED_DATE_TIME =
+            new AvroInstantDeserializer<>(ZonedDateTime.class, ZonedDateTime::ofInstant);
+
+    protected final BiFunction<Instant, ZoneId, T> fromInstant;
+
+    protected AvroInstantDeserializer(Class<T> t, BiFunction<Instant, ZoneId, T> fromInstant) {
+        super(t);
+        this.fromInstant = fromInstant;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public T deserialize(JsonParser p, DeserializationContext context) throws IOException, JsonProcessingException {
+        final ZoneId defaultZoneId = context.getTimeZone().toZoneId().normalized();
+        switch (p.getCurrentToken()) {
+            case VALUE_NUMBER_INT:
+                return fromLong(p.getLongValue(), defaultZoneId);
+            default:
+                try {
+                    return (T) context.handleUnexpectedToken(_valueClass, p);
+                } catch (JsonMappingException e) {
+                    throw e;
+                } catch (IOException e) {
+                    throw JsonMappingException.fromUnexpectedIOE(e);
+                }
+        }
+    }
+
+    @Override
+    public JsonDeserializer<T> createContextual(DeserializationContext ctxt, BeanProperty property) {
+        return this;
+    }
+
+    private T fromLong(long longValue, ZoneId defaultZoneId) {
+        return fromInstant.apply(Instant.ofEpochMilli(longValue), defaultZoneId);
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/AvroInstantSerializer.java

@@ -0,0 +1,72 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.BeanProperty;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonFormatVisitorWrapper;
+import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonIntegerFormatVisitor;
+import com.fasterxml.jackson.databind.ser.ContextualSerializer;
+import com.fasterxml.jackson.databind.ser.std.StdScalarSerializer;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+import java.time.temporal.Temporal;
+import java.util.function.Function;
+
+/**
+ * A serializer for variants of java.time classes that represent a specific instant on the timeline
+ * (Instant, OffsetDateTime, ZonedDateTime) which supports serialization to Avro long type and logicalType.
+ *
+ * See: http://avro.apache.org/docs/current/spec.html#Logical+Types
+ *
+ * Note: {@link AvroInstantSerializer} does not support serialization to string.
+ *
+ * @param <T> The type of a instant class that can be serialized.
+ */
+public class AvroInstantSerializer<T extends Temporal> extends StdScalarSerializer<T>
+        implements ContextualSerializer {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final AvroInstantSerializer<Instant> INSTANT =
+            new AvroInstantSerializer<>(Instant.class, Function.identity());
+
+    public static final AvroInstantSerializer<OffsetDateTime> OFFSET_DATE_TIME =
+            new AvroInstantSerializer<>(OffsetDateTime.class, OffsetDateTime::toInstant);
+
+    public static final AvroInstantSerializer<ZonedDateTime> ZONED_DATE_TIME =
+            new AvroInstantSerializer<>(ZonedDateTime.class, ZonedDateTime::toInstant);
+
+    private final Function<T, Instant> getInstant;
+
+    protected AvroInstantSerializer(Class<T> t, Function<T, Instant> getInstant) {
+        super(t);
+        this.getInstant = getInstant;
+    }
+
+    @Override
+    public void serialize(T value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+        final Instant instant = getInstant.apply(value);
+        gen.writeNumber(instant.toEpochMilli());
+    }
+
+    @Override
+    public JsonSerializer<?> createContextual(SerializerProvider prov, BeanProperty property) {
+        return this;
+    }
+
+    @Override
+    public void acceptJsonFormatVisitor(JsonFormatVisitorWrapper visitor, JavaType typeHint) throws JsonMappingException {
+        JsonIntegerFormatVisitor v2 = visitor.expectIntegerFormat(typeHint);
+        if (v2 != null) {
+            v2.numberType(JsonParser.NumberType.LONG);
+        }
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/AvroJavaTimeModule.java

@@ -0,0 +1,43 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310;
+
+import com.fasterxml.jackson.core.Version;
+import com.fasterxml.jackson.core.json.PackageVersion;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+
+/**
+ * A module that installs a collection of serializers and deserializers for java.time classes.
+ */
+public class AvroJavaTimeModule extends SimpleModule {
+
+    private static final long serialVersionUID = 1L;
+
+    public AvroJavaTimeModule() {
+        super(PackageVersion.VERSION);
+        addSerializer(Instant.class, AvroInstantSerializer.INSTANT);
+        addSerializer(OffsetDateTime.class, AvroInstantSerializer.OFFSET_DATE_TIME);
+        addSerializer(ZonedDateTime.class, AvroInstantSerializer.ZONED_DATE_TIME);
+
+        addDeserializer(Instant.class, AvroInstantDeserializer.INSTANT);
+        addDeserializer(OffsetDateTime.class, AvroInstantDeserializer.OFFSET_DATE_TIME);
+        addDeserializer(ZonedDateTime.class, AvroInstantDeserializer.ZONED_DATE_TIME);
+    }
+
+    @Override
+    public String getModuleName() {
+        return getClass().getName();
+    }
+
+    @Override
+    public Version version() {
+        return PackageVersion.VERSION;
+    }
+
+    @Override
+    public void setupModule(SetupContext context) {
+        super.setupModule(context);
+    }
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/DateTimeVisitor.java

@@ -0,0 +1,87 @@
+package com.fasterxml.jackson.dataformat.avro.schema;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonIntegerFormatVisitor;
+import org.apache.avro.LogicalType;
+import org.apache.avro.Schema;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+import java.util.Date;
+
+public class DateTimeVisitor extends JsonIntegerFormatVisitor.Base
+        implements SchemaBuilder {
+
+    protected JsonParser.NumberType _type;
+    protected JavaType _hint;
+
+    public DateTimeVisitor() {
+    }
+
+    public DateTimeVisitor(JavaType typeHint) {
+        _hint = typeHint;
+    }
+
+    @Override
+    public void numberType(JsonParser.NumberType type) {
+        _type = type;
+    }
+
+    @Override
+    public Schema builtAvroSchema() {
+        if (_type == null) {
+            throw new IllegalStateException("No number type indicated");
+        }
+
+        Schema schema = AvroSchemaHelper.numericAvroSchema(_type);
+        if (_hint != null) {
+            String logicalType = logicalType(_hint);
+            if (logicalType != null) {
+                schema.addProp(LogicalType.LOGICAL_TYPE_PROP, logicalType);
+            } else {
+                schema.addProp(AvroSchemaHelper.AVRO_SCHEMA_PROP_CLASS, AvroSchemaHelper.getTypeId(_hint));
+            }
+        }
+        return schema;
+    }
+
+    private String logicalType(JavaType hint) {
+        Class<?> clazz = hint.getRawClass();
+
+        if (Date.class.isAssignableFrom(clazz)) {
+            return TIMESTAMP_MILLIS;
+        }
+        if (OffsetDateTime.class.isAssignableFrom(clazz)) {
+            return TIMESTAMP_MILLIS;
+        }
+        if (ZonedDateTime.class.isAssignableFrom(clazz)) {
+            return TIMESTAMP_MILLIS;
+        }
+        if (Instant.class.isAssignableFrom(clazz)) {
+            return TIMESTAMP_MILLIS;
+        }
+
+        if (LocalDate.class.isAssignableFrom(clazz)) {
+            return DATE;
+        }
+        if (LocalTime.class.isAssignableFrom(clazz)) {
+            return TIME_MILLIS;
+        }
+        if (LocalDateTime.class.isAssignableFrom(clazz)) {
+            return LOCAL_TIMESTAMP_MILLIS;
+        }
+
+        return null;
+    }
+
+    private static final String DATE = "date";
+    private static final String TIME_MILLIS = "time-millis";
+    private static final String TIMESTAMP_MILLIS = "timestamp-millis";
+    private static final String LOCAL_TIMESTAMP_MILLIS = "local-timestamp-millis";
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/VisitorFormatWrapperImpl.java

@@ -11,11 +11,14 @@
 import com.fasterxml.jackson.databind.jsonFormatVisitors.*;
 import com.fasterxml.jackson.dataformat.avro.AvroSchema;
 
+import java.time.temporal.Temporal;
+import java.util.Date;
+
 public class VisitorFormatWrapperImpl
     implements JsonFormatVisitorWrapper
 {
     protected SerializerProvider _provider;
-    
+
     protected final DefinedSchemas _schemas;
 
     /**
@@ -28,7 +31,7 @@ public class VisitorFormatWrapperImpl
      * Schema for simple types that do not need a visitor.
      */
     protected Schema _valueSchema;
-    
+
     /*
     /**********************************************************************
     /* Construction
@@ -39,7 +42,7 @@ public VisitorFormatWrapperImpl(DefinedSchemas schemas, SerializerProvider p) {
         _schemas = schemas;
         _provider = p;
     }
-    
+
     @Override
     public SerializerProvider getProvider() {
         return _provider;
@@ -67,7 +70,7 @@ public Schema getAvroSchema() {
         }
         return _builder.builtAvroSchema();
     }
-    
+
     /*
     /**********************************************************************
     /* Callbacks
@@ -97,7 +100,7 @@ public JsonMapFormatVisitor expectMapFormat(JavaType mapType) {
         _builder = v;
         return v;
     }
-    
+
     @Override
     public JsonArrayFormatVisitor expectArrayFormat(final JavaType convertedType) {
         // 22-Mar-2016, tatu: Actually we can detect byte[] quite easily here can't we?
@@ -148,6 +151,13 @@ public JsonIntegerFormatVisitor expectIntegerFormat(JavaType type) {
             _valueSchema = s;
             return null;
         }
+
+        if (isDateTimeType(type)) {
+            DateTimeVisitor v = new DateTimeVisitor(type);
+            _builder = v;
+            return v;
+        }
+
         IntegerVisitor v = new IntegerVisitor(type);
         _builder = v;
         return v;
@@ -186,4 +196,14 @@ protected <T> T _throwUnsupported() {
     protected <T> T _throwUnsupported(String msg) {
         throw new UnsupportedOperationException(msg);
     }
+
+    private boolean isDateTimeType(JavaType type) {
+        if (Temporal.class.isAssignableFrom(type.getRawClass())) {
+            return true;
+        }
+        if (Date.class.isAssignableFrom(type.getRawClass())) {
+            return true;
+        }
+        return false;
+    }
 }