AVRO-2123: Java duration logical type (#2520)

This adds a new `java.time.TemporalAmount` implementation, which
supports the Avro `duration` logical type (neither `java.time.Period`
nor `java.time.Duration` support it).

This type is then used in the conversion for the (new) logical type
implementation "duration".

Last, the logical type "uuid" is refactored to include validation.

lang/java/avro/src/main/java/org/apache/avro/Conversions.java

@@ -18,15 +18,18 @@
 
 package org.apache.avro;
 
-import java.math.RoundingMode;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericEnumSymbol;
 import org.apache.avro.generic.GenericFixed;
 import org.apache.avro.generic.IndexedRecord;
+import org.apache.avro.util.TimePeriod;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.math.RoundingMode;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.IntBuffer;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Map;
@@ -147,6 +150,43 @@ private static BigDecimal validate(final LogicalTypes.Decimal decimal, BigDecima
     }
   }
 
+  public static class DurationConversion extends Conversion<TimePeriod> {
+    @Override
+    public Class<TimePeriod> getConvertedType() {
+      return TimePeriod.class;
+    }
+
+    @Override
+    public String getLogicalTypeName() {
+      return "duration";
+    }
+
+    @Override
+    public Schema getRecommendedSchema() {
+      return LogicalTypes.duration().addToSchema(Schema.createFixed("time.Duration",
+          "A 12-byte byte array encoding a duration in months, days and milliseconds.", null, 12));
+    }
+
+    @Override
+    public TimePeriod fromFixed(GenericFixed value, Schema schema, LogicalType type) {
+      IntBuffer buffer = ByteBuffer.wrap(value.bytes()).order(ByteOrder.LITTLE_ENDIAN).asIntBuffer();
+      long months = Integer.toUnsignedLong(buffer.get());
+      long days = Integer.toUnsignedLong(buffer.get());
+      long millis = Integer.toUnsignedLong(buffer.get());
+      return TimePeriod.of(months, days, millis);
+    }
+
+    @Override
+    public GenericFixed toFixed(TimePeriod value, Schema schema, LogicalType type) {
+      ByteBuffer buffer = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN);
+      IntBuffer intBuffer = buffer.asIntBuffer();
+      intBuffer.put((int) value.getMonths());
+      intBuffer.put((int) value.getDays());
+      intBuffer.put((int) value.getMillis());
+      return new GenericData.Fixed(schema, buffer.array());
+    }
+  }
+
   /**
    * Convert an underlying representation of a logical type (such as a ByteBuffer)
    * to a higher level object (such as a BigDecimal).

lang/java/avro/src/main/java/org/apache/avro/LogicalTypes.java

@@ -18,15 +18,15 @@
 
 package org.apache.avro;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 import java.util.ServiceLoader;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 public class LogicalTypes {
 
   private static final Logger LOG = LoggerFactory.getLogger(LogicalTypes.class);
@@ -182,6 +182,7 @@ private static LogicalType fromSchemaImpl(Schema schema, boolean throwErrors) {
   }
 
   private static final String DECIMAL = "decimal";
+  private static final String DURATION = "duration";
   private static final String UUID = "uuid";
   private static final String DATE = "date";
   private static final String TIME_MILLIS = "time-millis";
@@ -201,12 +202,18 @@ public static Decimal decimal(int precision, int scale) {
     return new Decimal(precision, scale);
   }
 
-  private static final LogicalType UUID_TYPE = new LogicalType("uuid");
+  private static final LogicalType UUID_TYPE = new Uuid();
 
   public static LogicalType uuid() {
     return UUID_TYPE;
   }
 
+  private static final LogicalType DURATION_TYPE = new Duration();
+
+  public static LogicalType duration() {
+    return DURATION_TYPE;
+  }
+
   private static final Date DATE_TYPE = new Date();
 
   public static Date date() {
@@ -249,6 +256,38 @@ public static LocalTimestampMicros localTimestampMicros() {
     return LOCAL_TIMESTAMP_MICROS_TYPE;
   }
 
+  /** Uuid represents a uuid without a time */
+  public static class Uuid extends LogicalType {
+    private Uuid() {
+      super(UUID);
+    }
+
+    @Override
+    public void validate(Schema schema) {
+      super.validate(schema);
+      if (schema.getType() != Schema.Type.STRING) {
+        throw new IllegalArgumentException("Uuid can only be used with an underlying string type");
+      }
+    }
+  }
+
+  /**
+   * Duration represents a duration, consisting on months, days and milliseconds
+   */
+  public static class Duration extends LogicalType {
+    private Duration() {
+      super(DURATION);
+    }
+
+    @Override
+    public void validate(Schema schema) {
+      super.validate(schema);
+      if (schema.getType() != Schema.Type.FIXED || schema.getFixedSize() != 12) {
+        throw new IllegalArgumentException("Duration can only be used with an underlying fixed type of size 12.");
+      }
+    }
+  }
+
   /** Decimal represents arbitrary-precision fixed-scale decimal numbers */
   public static class Decimal extends LogicalType {
     private static final String PRECISION_PROP = "precision";