AVRO-3644: [java] Support java.util.Optional in reflect package

lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java

@@ -25,6 +25,7 @@
 import java.util.HashMap;
 import java.util.Collection;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Conversion;
@@ -297,6 +298,15 @@ protected void readField(Object record, Field field, Object oldDatum, ResolvingD
             return;
           }
         }
+        if (Optional.class.isAssignableFrom(accessor.getField().getType())) {
+          try {
+            Object value = readWithoutConversion(oldDatum, field.schema(), in);
+            accessor.set(record, Optional.ofNullable(value));
+            return;
+          } catch (IllegalAccessException e) {
+            throw new AvroRuntimeException("Failed to set " + field);
+          }
+        }
         try {
           accessor.set(record, readWithoutConversion(oldDatum, field.schema(), in));
           return;

lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumWriter.java

@@ -22,6 +22,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.avro.AvroRuntimeException;
@@ -153,6 +154,8 @@ else if (datum instanceof Map && ReflectData.isNonStringMapSchema(schema)) {
         entryList.add(new MapEntry(e.getKey(), e.getValue()));
       }
       datum = entryList;
+    } else if (datum instanceof Optional) {
+      datum = ((Optional) datum).orElse(null);
     }
     try {
       super.write(schema, datum, out);

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificData.java

@@ -47,6 +47,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -384,6 +385,8 @@ else if (type instanceof ParameterizedType) {
         if (!(key instanceof Class && CharSequence.class.isAssignableFrom((Class<?>) key)))
           throw new AvroTypeException("Map key class not CharSequence: " + SchemaUtil.describe(key));
         return Schema.createMap(createSchema(value, names));
+      } else if (Optional.class.isAssignableFrom(raw)) {
+        return Schema.createUnion(Schema.create(Schema.Type.NULL), createSchema(params[0], names));
       } else {
         return createSchema(raw, names);
       }

lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflect.java

@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Random;
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.AvroTypeException;
@@ -1281,6 +1282,18 @@ public void testMultipleFieldAliases() {
     check(ClassWithMultipleAliasesOnField.class, expectedSchema.toString());
   }
 
+  private static class OptionalTest {
+    Optional<Integer> foo;
+  }
+
+  @Test
+  public void testOptional() {
+    check(OptionalTest.class,
+        "{\"type\":\"record\",\"name\":\"OptionalTest\","
+            + "\"namespace\":\"org.apache.avro.reflect.TestReflect\",\"fields\":["
+            + "{\"name\":\"foo\",\"type\":[\"null\",\"int\"],\"default\":null}]}");
+  }
+
   private static class DefaultTest {
     @AvroDefault("1")
     int foo;

lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.avro.io.Decoder;
 import org.apache.avro.io.DecoderFactory;
@@ -125,6 +126,40 @@ public void testRead_PojoWithMap() throws IOException {
     assertEquals(pojoWithMap, deserialized);
   }
 
+  @Test
+  public void testRead_PojoWithOptional() throws IOException {
+    PojoWithOptional pojoWithOptional = new PojoWithOptional();
+    pojoWithOptional.setId(42);
+    pojoWithOptional.setRelatedId(Optional.of(13));
+
+    byte[] serializedBytes = serializeWithReflectDatumWriter(pojoWithOptional, PojoWithOptional.class);
+
+    Decoder decoder = DecoderFactory.get().binaryDecoder(serializedBytes, null);
+    ReflectDatumReader<PojoWithOptional> reflectDatumReader = new ReflectDatumReader<>(PojoWithOptional.class);
+
+    PojoWithOptional deserialized = new PojoWithOptional();
+    reflectDatumReader.read(deserialized, decoder);
+
+    assertEquals(pojoWithOptional, deserialized);
+  }
+
+  @Test
+  public void testRead_PojoWithEmptyOptional() throws IOException {
+    PojoWithOptional pojoWithOptional = new PojoWithOptional();
+    pojoWithOptional.setId(42);
+    pojoWithOptional.setRelatedId(Optional.empty());
+
+    byte[] serializedBytes = serializeWithReflectDatumWriter(pojoWithOptional, PojoWithOptional.class);
+
+    Decoder decoder = DecoderFactory.get().binaryDecoder(serializedBytes, null);
+    ReflectDatumReader<PojoWithOptional> reflectDatumReader = new ReflectDatumReader<>(PojoWithOptional.class);
+
+    PojoWithOptional deserialized = new PojoWithOptional();
+    reflectDatumReader.read(deserialized, decoder);
+
+    assertEquals(pojoWithOptional, deserialized);
+  }
+
   public static class PojoWithList {
     private int id;
     private List<Integer> relatedIds;
@@ -309,4 +344,52 @@ public boolean equals(Object obj) {
         return relatedIds.equals(other.relatedIds);
     }
   }
+
+  public static class PojoWithOptional {
+    private int id;
+
+    private Optional<Integer> relatedId;
+
+    public int getId() {
+      return id;
+    }
+
+    public void setId(int id) {
+      this.id = id;
+    }
+
+    public Optional<Integer> getRelatedId() {
+      return relatedId;
+    }
+
+    public void setRelatedId(Optional<Integer> relatedId) {
+      this.relatedId = relatedId;
+    }
+
+    @Override
+    public int hashCode() {
+      final int prime = 31;
+      int result = 1;
+      result = prime * result + id;
+      result = prime * result + ((relatedId == null) ? 0 : relatedId.hashCode());
+      return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj)
+        return true;
+      if (obj == null)
+        return false;
+      if (getClass() != obj.getClass())
+        return false;
+      PojoWithOptional other = (PojoWithOptional) obj;
+      if (id != other.id)
+        return false;
+      if (relatedId == null) {
+        return other.relatedId == null;
+      } else
+        return relatedId.equals(other.relatedId);
+    }
+  }
 }