[Avro] logicalType support for some java.time types. (#283)

Add

* `logicalType` schema support for few date time types.
* Support to serialize and de-serialize java.time types into Avro type and logicalType.

avro/README.md

@@ -111,6 +111,40 @@ byte[] avroData = mapper.writer(schema)
 
 and that's about it, for now.
 
+## Java Time Support
+Serialization and deserialization support for limited set of `java.time` classes to Avro with [logical type](http://avro.apache.org/docs/current/spec.html#Logical+Types) is provided by `AvroJavaTimeModule`.
+
+This module is to be used either:
+- Instead of Java 8 date/time module (`com.fasterxml.jackson.datatype.jsr310.JavaTimeModule`) or
+- to override Java 8 date/time module and for that, module must be registered AFTER Java 8 date/time module (last registration wins).
+
+```java
+AvroMapper mapper = AvroMapper.builder()
+    .addModule(new AvroJavaTimeModule())
+    .build();
+```
+
+#### Note
+Please note that time zone information is lost at serialization. Serialized values represent point in time, 
+independent of a particular time zone or calendar. Upon reading a value back time instant is reconstructed but not the original time zone.
+
+#### Supported java.time types:
+
+Supported java.time types with Avro schema.  
+
+| Type                           | Avro schema
+| ------------------------------ | -------------
+| `java.time.OffsetDateTime`     | `{"type": "long", "logicalType": "timestamp-millis"}`
+| `java.time.ZonedDateTime`      | `{"type": "long", "logicalType": "timestamp-millis"}`
+| `java.time.Instant`            | `{"type": "long", "logicalType": "timestamp-millis"}`
+| `java.time.LocalDate`          | `{"type": "int",  "logicalType": "date"}`
+| `java.time.LocalTime`          | `{"type": "int",  "logicalType": "time-millis"}`
+| `java.time.LocalDateTime`      | `{"type": "long", "logicalType": "local-timestamp-millis"}`
+
+#### Precision
+
+Avro supports milliseconds and microseconds precision for date and time related LogicalTypes, but this module only supports millisecond precision.
+
 ## Generating Avro Schema from POJO definition
 
 Ok but wait -- you do not have to START with an Avro Schema. This module can

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

@@ -0,0 +1,50 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310;
+
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.dataformat.avro.PackageVersion;
+import com.fasterxml.jackson.dataformat.avro.jsr310.deser.AvroInstantDeserializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.deser.AvroLocalDateDeserializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.deser.AvroLocalDateTimeDeserializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.deser.AvroLocalTimeDeserializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.ser.AvroInstantSerializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.ser.AvroLocalDateSerializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.ser.AvroLocalDateTimeSerializer;
+import com.fasterxml.jackson.dataformat.avro.jsr310.ser.AvroLocalTimeSerializer;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+
+/**
+ * A module that installs a collection of serializers and deserializers for java.time classes.
+ *
+ * This module is to be used either:
+ *   - Instead of Java 8 date/time module (com.fasterxml.jackson.datatype.jsr310.JavaTimeModule) or
+ *   - to override Java 8 date/time module and for that, module must be registered AFTER Java 8 date/time module.
+ */
+public class AvroJavaTimeModule extends SimpleModule {
+
+    private static final long serialVersionUID = 1L;
+
+    public AvroJavaTimeModule() {
+        super(AvroJavaTimeModule.class.getName(), PackageVersion.VERSION);
+
+        addSerializer(Instant.class, AvroInstantSerializer.INSTANT);
+        addSerializer(OffsetDateTime.class, AvroInstantSerializer.OFFSET_DATE_TIME);
+        addSerializer(ZonedDateTime.class, AvroInstantSerializer.ZONED_DATE_TIME);
+        addSerializer(LocalDateTime.class, AvroLocalDateTimeSerializer.INSTANCE);
+        addSerializer(LocalDate.class, AvroLocalDateSerializer.INSTANCE);
+        addSerializer(LocalTime.class, AvroLocalTimeSerializer.INSTANCE);
+
+        addDeserializer(Instant.class, AvroInstantDeserializer.INSTANT);
+        addDeserializer(OffsetDateTime.class, AvroInstantDeserializer.OFFSET_DATE_TIME);
+        addDeserializer(ZonedDateTime.class, AvroInstantDeserializer.ZONED_DATE_TIME);
+        addDeserializer(LocalDateTime.class, AvroLocalDateTimeDeserializer.INSTANCE);
+        addDeserializer(LocalDate.class, AvroLocalDateDeserializer.INSTANCE);
+        addDeserializer(LocalTime.class, AvroLocalTimeDeserializer.INSTANCE);
+    }
+
+}

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

@@ -0,0 +1,50 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.deser;
+
+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;
+
+/**
+ * Deserializer for variants of java.time classes (Instant, OffsetDateTime, ZonedDateTime) from an integer value.
+ *
+ * Deserialized value represents an instant on the global timeline, independent of a particular time zone or
+ * calendar, with a precision of one millisecond from the unix epoch, 1 January 1970 00:00:00.000 UTC.
+ * Time zone information is lost at serialization. Time zone data types receives time zone from deserialization context.
+ *
+ * Deserialization from string is not supported.
+ *
+ * @param <T> The type of a instant class that can be deserialized.
+ */
+public class AvroInstantDeserializer<T extends Temporal> extends AvroJavaTimeDeserializerBase <T> {
+
+    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> supportedType, BiFunction<Instant, ZoneId, T> fromInstant) {
+        super(supportedType);
+        this.fromInstant = fromInstant;
+    }
+
+    @Override
+    protected T fromLong(long longValue, ZoneId defaultZoneId) {
+        /**
+         * Number of milliseconds, independent of a particular time zone or calendar,
+         * from 1 January 1970 00:00:00.000 UTC.
+         */
+        return fromInstant.apply(Instant.ofEpochMilli(longValue), defaultZoneId);
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/deser/AvroJavaTimeDeserializerBase.java

@@ -0,0 +1,36 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.deser;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.deser.std.StdScalarDeserializer;
+import com.fasterxml.jackson.databind.type.LogicalType;
+
+import java.io.IOException;
+import java.time.ZoneId;
+
+import static com.fasterxml.jackson.core.JsonToken.VALUE_NUMBER_INT;
+
+public abstract class AvroJavaTimeDeserializerBase<T> extends StdScalarDeserializer<T> {
+
+    protected AvroJavaTimeDeserializerBase(Class<T> supportedType) {
+        super(supportedType);
+    }
+
+    @Override
+    public LogicalType logicalType() {
+        return LogicalType.DateTime;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public T deserialize(JsonParser p, DeserializationContext context) throws IOException {
+        if (p.getCurrentToken() == VALUE_NUMBER_INT) {
+            final ZoneId defaultZoneId = context.getTimeZone().toZoneId().normalized();
+            return fromLong(p.getLongValue(), defaultZoneId);
+        } else {
+            return (T) context.handleUnexpectedToken(_valueClass, p);
+        }
+    }
+
+    protected abstract T fromLong(long longValue, ZoneId defaultZoneId);
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/deser/AvroLocalDateDeserializer.java

@@ -0,0 +1,31 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.deser;
+
+import java.time.LocalDate;
+import java.time.ZoneId;
+
+/**
+ * Deserializer for {@link LocalDate} from and integer value.
+ *
+ * Deserialized value represents number of days from the unix epoch, 1 January 1970.
+ *
+ * Deserialization from string is not supported.
+ */
+public class AvroLocalDateDeserializer extends AvroJavaTimeDeserializerBase<LocalDate> {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final AvroLocalDateDeserializer INSTANCE = new AvroLocalDateDeserializer();
+
+    protected AvroLocalDateDeserializer() {
+        super(LocalDate.class);
+    }
+
+    @Override
+    protected LocalDate fromLong(long longValue, ZoneId defaultZoneId) {
+        /**
+         * Number of days from the unix epoch, 1 January 1970..
+         */
+        return LocalDate.ofEpochDay(longValue);
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/deser/AvroLocalDateTimeDeserializer.java

@@ -0,0 +1,35 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.deser;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+
+/**
+ * Deserializer for {@link LocalDateTime} from an integer value.
+ *
+ * Deserialized value represents timestamp in a local timezone, regardless of what specific time zone
+ * is considered local, with a precision of one millisecond from 1 January 1970 00:00:00.000.
+ *
+ * Deserialization from string is not supported.
+ */
+public class AvroLocalDateTimeDeserializer extends AvroJavaTimeDeserializerBase<LocalDateTime> {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final AvroLocalDateTimeDeserializer INSTANCE = new AvroLocalDateTimeDeserializer();
+
+    protected AvroLocalDateTimeDeserializer() {
+        super(LocalDateTime.class);
+    }
+
+    @Override
+    protected LocalDateTime fromLong(long longValue, ZoneId defaultZoneId) {
+        /**
+         * Number of milliseconds in a local timezone, regardless of what specific time zone is considered local,
+         * from 1 January 1970 00:00:00.000.
+         */
+        return LocalDateTime.ofInstant(Instant.ofEpochMilli(longValue), ZoneOffset.ofTotalSeconds(0));
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/jsr310/deser/AvroLocalTimeDeserializer.java

@@ -0,0 +1,33 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.deser;
+
+import java.time.LocalTime;
+import java.time.ZoneId;
+
+/**
+ * Deserializer for {@link LocalTime} from an integer value.
+ *
+ * Deserialized value represents time of day, with no reference to a particular calendar,
+ * time zone or date, where the int stores the number of milliseconds after midnight, 00:00:00.000.
+ *
+ * Deserialization from string is not supported.
+ */
+public class AvroLocalTimeDeserializer extends AvroJavaTimeDeserializerBase<LocalTime> {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final AvroLocalTimeDeserializer INSTANCE = new AvroLocalTimeDeserializer();
+
+    protected AvroLocalTimeDeserializer() {
+        super(LocalTime.class);
+    }
+
+    @Override
+    protected LocalTime fromLong(long longValue, ZoneId defaultZoneId) {
+        /**
+         * Number of milliseconds, with no reference to a particular calendar, time zone or date, after
+         * midnight, 00:00:00.000.
+         */
+        return LocalTime.ofNanoOfDay(longValue * 1000_000L);
+    }
+
+}

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

@@ -0,0 +1,76 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310.ser;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.JsonMappingException;
+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.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;
+
+/**
+ * Serializer for variants of java.time classes (Instant, OffsetDateTime, ZonedDateTime) into long value.
+ *
+ * Serialized value represents an instant on the global timeline, independent of a particular time zone or
+ * calendar, with a precision of one millisecond from the unix epoch, 1 January 1970 00:00:00.000 UTC.
+ * Please note that time zone information gets lost in this process. Upon reading a value back, we can only
+ * reconstruct the instant, but not the original representation.
+ *
+ * Note: In combination with {@link com.fasterxml.jackson.dataformat.avro.schema.DateTimeVisitor} it aims to produce
+ * Avro schema with type long and logicalType timestamp-millis:
+ * {
+ *   "type" : "long",
+ *   "logicalType" : "timestamp-millis"
+ * }
+ *
+ * {@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> {
+
+    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 {
+        /**
+         * Number of milliseconds, independent of a particular time zone or calendar,
+         * from 1 January 1970 00:00:00.000 UTC.
+         */
+        final Instant instant = getInstant.apply(value);
+        gen.writeNumber(instant.toEpochMilli());
+    }
+
+    @Override
+    public void acceptJsonFormatVisitor(JsonFormatVisitorWrapper visitor, JavaType typeHint) throws JsonMappingException {
+        JsonIntegerFormatVisitor v2 = visitor.expectIntegerFormat(typeHint);
+        if (v2 != null) {
+            v2.numberType(JsonParser.NumberType.LONG);
+        }
+    }
+
+}