AVRO-2639: add option to write Optional only when fields are nullable

lang/java/compiler/src/main/java/org/apache/avro/compiler/specific/SpecificCompiler.java

@@ -113,6 +113,7 @@ void addLogicalTypeConversions(SpecificData specificData) {
   private FieldVisibility fieldVisibility = FieldVisibility.PRIVATE;
   private boolean createOptionalGetters = false;
   private boolean gettersReturnOptional = false;
+  private boolean gettersReturnOptionalOnlyForNullable = false;
   private boolean createSetters = true;
   private boolean createAllArgsConstructor = true;
   private String outputCharacterEncoding;
@@ -238,7 +239,8 @@ public boolean isCreateOptionalGetters() {
   }
 
   /**
-   * Set to false to not create the getters that return an Optional.
+   * Set to true to create getters of the form
+   * {@code getOptionalFoo<Optional<T>>()}.=
    */
   public void setCreateOptionalGetters(boolean createOptionalGetters) {
     this.createOptionalGetters = createOptionalGetters;
@@ -249,12 +251,27 @@ public boolean isGettersReturnOptional() {
   }
 
   /**
-   * Set to false to not create the getters that return an Optional.
+   * Set to to true to make the regular getters return Optional. Generated code
+   * will be {@code getFoo<Optional<T>>()}
    */
   public void setGettersReturnOptional(boolean gettersReturnOptional) {
     this.gettersReturnOptional = gettersReturnOptional;
   }
 
+  public boolean isGettersReturnOptionalOnlyForNullable() {
+    return this.gettersReturnOptionalOnlyForNullable;
+  }
+
+  /**
+   * Set to true to make getters return Optional only if the underlying field in
+   * nullable. Generated code will be either {@code getFoo<T>()} or
+   * {@code getFoo<Optional<T>>()}. This setting only takes effect if
+   * {@link #gettersReturnOptional} is true.
+   */
+  public void setGettersReturnOptionalOnlyForNullable(boolean gettersReturnOptionalOnlyForNullable) {
+    this.gettersReturnOptionalOnlyForNullable = gettersReturnOptionalOnlyForNullable;
+  }
+
   /**
    * Set to true to use {@link java.math.BigDecimal} instead of
    * {@link java.nio.ByteBuffer} for logical type "decimal"

lang/java/compiler/src/main/velocity/org/apache/avro/compiler/specific/templates/java/classic/record.vm

@@ -209,7 +209,7 @@ static {
   }
 
 #foreach ($field in $schema.getFields())
-#if (${this.gettersReturnOptional})
+#if (${this.gettersReturnOptional} && ((${this.gettersReturnOptionalOnlyForNullable} && ${field.schema().isNullable()}) || !${this.gettersReturnOptionalOnlyForNullable}))
   /**
    * Gets the value of the '${this.mangle($field.name(), $schema.isError())}' field as an Optional<${this.javaType($field.schema())}>.
 #if ($field.doc())      * $field.doc()

lang/java/compiler/src/test/java/org/apache/avro/compiler/specific/TestSpecificCompiler.java

@@ -698,6 +698,45 @@ public void testPojoWithOptionalCreatedWhenOptionTurnedOn() throws IOException {
     assertEquals(9, optionalFound);
   }
 
+  @Test
+  public void testPojoWithOptionalOnlyWhenNullableCreatedTurnedOn() throws IOException {
+    SpecificCompiler compiler = createCompiler();
+    compiler.setGettersReturnOptional(true);
+    compiler.setGettersReturnOptionalOnlyForNullable(true);
+    compiler.compileToDestination(this.src, OUTPUT_DIR.getRoot());
+    assertTrue(this.outputFile.exists());
+    int optionalFound = 0;
+    try (BufferedReader reader = new BufferedReader(new FileReader(this.outputFile))) {
+      String line;
+      while ((line = reader.readLine()) != null) {
+        line = line.trim();
+        if (line.contains("Optional")) {
+          optionalFound++;
+        }
+      }
+    }
+    // only the nullable field (four uses of optional) and the import should be
+    // found
+    assertEquals(5, optionalFound);
+  }
+
+  @Test
+  public void testPojoWithOptionalOnlyWhenNullableCreatedTurnedOnAndGettersReturnOptionalTurnedOff()
+      throws IOException {
+    SpecificCompiler compiler = createCompiler();
+    compiler.setGettersReturnOptionalOnlyForNullable(true);
+    compiler.compileToDestination(this.src, OUTPUT_DIR.getRoot());
+    assertTrue(this.outputFile.exists());
+    try (BufferedReader reader = new BufferedReader(new FileReader(this.outputFile))) {
+      String line;
+      while ((line = reader.readLine()) != null) {
+        line = line.trim();
+        // no optionals since gettersReturnOptionalOnlyForNullable is false
+        assertFalse(line.contains("Optional"));
+      }
+    }
+  }
+
   @Test
   public void testPojoWithOptionalCreatedWhenOptionalForEverythingTurnedOn() throws IOException {
     SpecificCompiler compiler = createCompiler();