[Avro] write enums as string

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/AvroSchemaGenerator.java

@@ -31,4 +31,17 @@ public AvroSchemaGenerator disableLogicalTypes() {
         super.disableLogicalTypes();
         return this;
     }
+
+    @Override
+    public AvroSchemaGenerator enableWriteEnumAsString() {
+        super.enableWriteEnumAsString();
+        return this;
+    }
+
+    @Override
+    public AvroSchemaGenerator disableWriteEnumAsString() {
+        super.disableWriteEnumAsString();
+        return this;
+    }
+
 }

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/EnumVisitor.java

@@ -0,0 +1,45 @@
+package com.fasterxml.jackson.dataformat.avro.schema;
+
+import com.fasterxml.jackson.databind.BeanDescription;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonStringFormatVisitor;
+
+import org.apache.avro.Schema;
+
+import java.util.ArrayList;
+import java.util.Set;
+
+public class EnumVisitor extends JsonStringFormatVisitor.Base
+    implements SchemaBuilder
+{
+    protected final SerializerProvider _provider;
+    protected final JavaType _type;
+    protected final DefinedSchemas _schemas;
+
+    protected Set<String> _enums;
+
+    public EnumVisitor(SerializerProvider provider, DefinedSchemas schemas, JavaType t) {
+        _schemas = schemas;
+        _type = t;
+        _provider = provider;
+    }
+
+    @Override
+    public void enumTypes(Set<String> enums) {
+        _enums = enums;
+    }
+
+    @Override
+    public Schema builtAvroSchema() {
+        if (_enums == null) {
+        	throw new IllegalStateException("Possible enum values cannot be null");
+        }
+
+        BeanDescription bean = _provider.getConfig().introspectClassAnnotations(_type);
+        Schema schema = AvroSchemaHelper.createEnumSchema(bean, new ArrayList<>(_enums));
+        _schemas.addSchema(_type, schema);
+        return schema;
+    }
+
+}

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/StringVisitor.java

@@ -1,6 +1,5 @@
 package com.fasterxml.jackson.dataformat.avro.schema;
 
-import java.util.ArrayList;
 import java.util.Set;
 
 import org.apache.avro.Schema;
@@ -18,13 +17,9 @@ public class StringVisitor extends JsonStringFormatVisitor.Base
 {
     protected final SerializerProvider _provider;
     protected final JavaType _type;
-    protected final DefinedSchemas _schemas;
 
-    protected Set<String> _enums;
-
-    public StringVisitor(SerializerProvider provider, DefinedSchemas schemas, JavaType t) {
-        _schemas = schemas;
-        _type = t;
+    public StringVisitor(SerializerProvider provider, JavaType type) {
+        _type = type;
         _provider = provider;
     }
 
@@ -35,7 +30,7 @@ public void format(JsonValueFormat format) {
 
     @Override
     public void enumTypes(Set<String> enums) {
-        _enums = enums;
+    	// Do nothing
     }
 
     @Override
@@ -50,11 +45,6 @@ public Schema builtAvroSchema() {
             return AvroSchemaHelper.createUUIDSchema();
         }
         BeanDescription bean = _provider.getConfig().introspectClassAnnotations(_type);
-        if (_enums != null) {
-            Schema s = AvroSchemaHelper.createEnumSchema(bean, new ArrayList<>(_enums));
-            _schemas.addSchema(_type, s);
-            return s;
-        }
         Schema schema = Schema.create(Schema.Type.STRING);
         // Stringable classes need to include the type
         if (AvroSchemaHelper.isStringable(bean.getClassInfo()) && !_type.hasRawClass(String.class)) {

avro/src/main/java/com/fasterxml/jackson/dataformat/avro/schema/VisitorFormatWrapperImpl.java

@@ -16,6 +16,7 @@
 import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonObjectFormatVisitor;
 import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonStringFormatVisitor;
 import com.fasterxml.jackson.dataformat.avro.AvroSchema;
+
 import org.apache.avro.Schema;
 
 import java.time.temporal.Temporal;
@@ -32,6 +33,11 @@ public class VisitorFormatWrapperImpl
      */
     protected boolean _logicalTypesEnabled = false;
 
+    /**
+     * @since 2.18
+     */
+    protected boolean _writeEnumAsString = false;
+
     /**
      * Visitor used for resolving actual Schema, if structured type
      * (or one with complex configuration)
@@ -123,6 +129,26 @@ public boolean isLogicalTypesEnabled() {
         return _logicalTypesEnabled;
     }
 
+    /**
+     * Enable Java enum to Avro string mapping.
+     */
+    public VisitorFormatWrapperImpl enableWriteEnumAsString() {
+    	_writeEnumAsString = true;
+        return this;
+    }
+
+    /**
+     * Disable Java enum to Avro string mapping.
+     */
+    public VisitorFormatWrapperImpl disableWriteEnumAsString() {
+        _writeEnumAsString = false;
+        return this;
+    }
+
+    public boolean isWriteEnumAsStringEnabled() {
+        return _writeEnumAsString;
+    }
+
     /*
     /**********************************************************************
     /* Callbacks
@@ -177,7 +203,14 @@ public JsonStringFormatVisitor expectStringFormat(JavaType type)
             _valueSchema = s;
             return null;
         }
-        StringVisitor v = new StringVisitor(_provider, _schemas, type);
+
+        if (Enum.class.isAssignableFrom(type.getRawClass()) && !isWriteEnumAsStringEnabled()) {
+        	EnumVisitor v = new EnumVisitor(_provider, _schemas, type);
+            _builder = v;
+            return v;
+        }
+
+        StringVisitor v = new StringVisitor(_provider, type);
         _builder = v;
         return v;
     }

avro/src/test/java/com/fasterxml/jackson/dataformat/avro/EnumTest.java

@@ -2,7 +2,8 @@
 
 public class EnumTest extends AvroTestBase
 {
-    protected final static String ENUM_SCHEMA_JSON = "{\n"
+    // gender as Avro enum
+	protected final static String ENUM_SCHEMA_JSON = "{\n"
             +"\"type\": \"record\",\n"
             +"\"name\": \"Employee\",\n"
             +"\"fields\": [\n"
@@ -11,6 +12,14 @@ public class EnumTest extends AvroTestBase
             +"}\n"
             +"]}";
 
+    // gender as Avro string
+    protected final static String STRING_SCHEMA_JSON = "{"
+            +" \"type\": \"record\", "
+            +" \"name\": \"Employee\", "
+            +" \"fields\": ["
+            +" {\"name\": \"gender\", \"type\": \"string\"}"
+            +"]}";
+
     protected enum Gender { M, F; }
 
     protected static class Employee {
@@ -23,7 +32,7 @@ protected static class EmployeeStr {
 
     private final AvroMapper MAPPER = newMapper();
 
-    public void testSimple() throws Exception
+    public void test_avroSchemaWithEnum_fromEnumValueToEnumValue() throws Exception
     {
         AvroSchema schema = MAPPER.schemaFrom(ENUM_SCHEMA_JSON);
         Employee input = new Employee();
@@ -40,7 +49,7 @@ public void testSimple() throws Exception
         assertEquals(Gender.F, output.gender);
     }
 
-    public void testEnumValueAsString() throws Exception
+    public void test_avroSchemaWithEnum_fromStringValueToEnumValue() throws Exception
     {
         AvroSchema schema = MAPPER.schemaFrom(ENUM_SCHEMA_JSON);
         EmployeeStr input = new EmployeeStr();
@@ -56,4 +65,42 @@ public void testEnumValueAsString() throws Exception
         assertNotNull(output);
         assertEquals(Gender.F, output.gender);
     }
+
+    public void test_avroSchemaWithString_fromEnumValueToEnumValue() throws Exception
+    {
+        AvroSchema schema = MAPPER.schemaFrom(STRING_SCHEMA_JSON);
+        Employee input = new Employee();
+        input.gender = Gender.F;
+
+        byte[] bytes = MAPPER.writer(schema).writeValueAsBytes(input);
+        assertNotNull(bytes);
+        // FIXME What is expected bytes length?
+//        assertEquals(1, bytes.length); // measured to be current exp size
+
+        // and then back
+        Employee output = MAPPER.readerFor(Employee.class).with(schema)
+                .readValue(bytes);
+        assertNotNull(output);
+        assertEquals(Gender.F, output.gender);
+    }
+
+    // Not sure this test makes sense
+    public void test_avroSchemaWithString_fromStringValueToEnumValue() throws Exception
+    {
+        AvroSchema schema = MAPPER.schemaFrom(STRING_SCHEMA_JSON);
+        EmployeeStr input = new EmployeeStr();
+        input.gender = "F";
+
+        byte[] bytes = MAPPER.writer(schema).writeValueAsBytes(input);
+        assertNotNull(bytes);
+        // FIXME What is expected bytes length?
+//        assertEquals(1, bytes.length); // measured to be current exp size
+
+        // and then back
+        Employee output = MAPPER.readerFor(Employee.class).with(schema)
+                .readValue(bytes);
+        assertNotNull(output);
+        assertEquals(Gender.F, output.gender);
+    }
+
 }

avro/src/test/java/com/fasterxml/jackson/dataformat/avro/schema/Enum_schemaCreationTest.java

@@ -0,0 +1,56 @@
+package com.fasterxml.jackson.dataformat.avro.schema;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.dataformat.avro.AvroMapper;
+import com.fasterxml.jackson.dataformat.avro.AvroTestBase;
+
+import org.apache.avro.Schema;
+import org.apache.avro.specific.SpecificData;
+import org.junit.Test;
+
+public class Enum_schemaCreationTest extends AvroTestBase {
+
+    static enum NumbersEnum {
+    	ONE, TWO, THREE
+    }
+
+    private final AvroMapper MAPPER = newMapper();
+
+    @Test
+    public void testJavaEnumToAvroEnum_test() throws JsonMappingException {
+        // GIVEN
+        AvroSchemaGenerator gen = new AvroSchemaGenerator();
+
+        // WHEN
+        MAPPER.acceptJsonFormatVisitor(NumbersEnum.class , gen);
+        Schema actualSchema = gen.getGeneratedSchema().getAvroSchema();
+
+        System.out.println("schema:\n" + actualSchema.toString(true));
+
+        // THEN
+        assertThat(actualSchema.getType()).isEqualTo( Schema.Type.ENUM);
+        assertThat(actualSchema.getEnumSymbols()).containsExactlyInAnyOrder("ONE", "TWO", "THREE");
+    }
+
+    @Test
+    public void testJavaEnumToAvroString_test() throws JsonMappingException {
+        // GIVEN
+        AvroSchemaGenerator gen = new AvroSchemaGenerator()
+        		.enableWriteEnumAsString();
+
+        // WHEN
+        MAPPER.acceptJsonFormatVisitor(NumbersEnum.class , gen);
+        Schema actualSchema = gen.getGeneratedSchema().getAvroSchema();
+
+        System.out.println("schema:\n" + actualSchema.toString(true));
+
+        // THEN
+        assertThat(actualSchema.getType()).isEqualTo( Schema.Type.STRING);
+
+        // When type is stringable then java-class property is addded.
+        assertThat(actualSchema.getProp(SpecificData.CLASS_PROP)).isNotEmpty();
+    }
+
+}