Support to serialize and de-serialize Instant, OffsetDateTime and Zon…

…edDateTime to Avro long type and logicalType.

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

@@ -0,0 +1,77 @@
+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 serializing to Avro long type and logicalType.
+ * <p>
+ * See: http://avro.apache.org/docs/current/spec.html#Logical+Types
+ * <p>
+ * 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;
+    }
+
+    protected AvroInstantSerializer(AvroInstantSerializer<T> base) {
+        super(base.handledType());
+        this.getInstant = base.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, IonTimestampInstantDeserializer.INSTANT);
+//        addDeserializer(OffsetDateTime.class, IonTimestampInstantDeserializer.OFFSET_DATE_TIME);
+//        addDeserializer(ZonedDateTime.class, IonTimestampInstantDeserializer.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/test/java/com/fasterxml/jackson/dataformat/avro/jsr310/AvroInstantSerializer_schemaCreationTest.java

@@ -0,0 +1,85 @@
+package com.fasterxml.jackson.dataformat.avro.jsr310;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.dataformat.avro.AvroMapper;
+import com.fasterxml.jackson.dataformat.avro.AvroSchema;
+import com.fasterxml.jackson.dataformat.avro.schema.AvroSchemaGenerator;
+import org.apache.avro.LogicalType;
+import org.apache.avro.Schema;
+import org.apache.avro.specific.SpecificData;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(Parameterized.class)
+public class AvroInstantSerializer_schemaCreationTest {
+
+    @Parameter
+    public Class testClass;
+
+    @Parameters(name = "{index}: With {0}")
+    public static Collection<Class> testData() {
+        return Arrays.asList(
+                WithInstant_TestClass.class,
+                WithOffsetDateTime_TestClass.class,
+                WithZonedDateTime_TestClass.class);
+    }
+
+    /**
+     * Note: Mandatory property does not have additional "NULL" type in Avro thus
+     * assert statements are simpler and can focus on relevant schema properties.
+     */
+    class WithInstant_TestClass {
+        @JsonProperty(required = true)
+        public Instant testedProperty;
+    }
+
+    class WithOffsetDateTime_TestClass {
+        @JsonProperty(required = true)
+        public OffsetDateTime testedProperty;
+    }
+
+    class WithZonedDateTime_TestClass {
+        @JsonProperty(required = true)
+        public ZonedDateTime testedProperty;
+    }
+
+    @Test
+    public void testSchemaCreation() throws JsonMappingException {
+        // GIVEN
+        AvroMapper mapper = AvroMapper.builder()
+                .addModules(new AvroJavaTimeModule())
+                .build();
+        AvroSchemaGenerator gen = new AvroSchemaGenerator();
+
+        // WHEN
+        mapper.acceptJsonFormatVisitor(testClass, gen);
+
+        // THEN
+        AvroSchema actualSchema = gen.getGeneratedSchema();
+        Schema avroSchema = actualSchema.getAvroSchema();
+        System.out.println(avroSchema.toString(true));
+
+        Schema.Field field = avroSchema.getField("testedProperty");
+        assertThat(field).isNotNull();
+        assertThat(field.schema().getType()).isEqualTo(Schema.Type.LONG);
+        assertThat(field.schema().getProp(LogicalType.LOGICAL_TYPE_PROP)).isEqualTo("timestamp-millis");
+        /**
+         * Having logicalType and java-class is not valid according to
+         * {@link org.apache.avro.LogicalType#validate(Schema)}
+         */
+        assertThat(field.schema().getProp(SpecificData.CLASS_PROP)).isNull();
+    }
+
+}