/*
 * Copyright 2019 Confluent Inc.
 * Portions Copyright 2020 Amazon.com, Inc. or its affiliates.
 * All Rights Reserved.
 * Licensed under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.amazonaws.services.schemaregistry.kafkaconnect.avrodata;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.IntNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.avro.JsonProperties;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericEnumSymbol;
import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.util.internal.JacksonUtils;
import org.apache.kafka.common.cache.Cache;
import org.apache.kafka.common.cache.LRUCache;
import org.apache.kafka.common.cache.SynchronizedCache;
import org.apache.kafka.connect.data.ConnectSchema;
import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.data.Time;
import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.errors.DataException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * Utilities for converting between our runtime data format and Avro, and (de)serializing that data.
 */
public class AvroData {

    private static final Logger log = LoggerFactory.getLogger(AvroData.class);

    public static final String NAMESPACE = "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata";
    // Avro does not permit empty schema names, which might be the ideal default since we also are
    // not permitted to simply omit the name. Instead, make it very clear where the default is
    // coming from.
    public static final String DEFAULT_SCHEMA_NAME = "ConnectDefault";
    public static final String DEFAULT_SCHEMA_FULL_NAME = NAMESPACE + "." + DEFAULT_SCHEMA_NAME;
    public static final String MAP_ENTRY_TYPE_NAME = "MapEntry";
    public static final String KEY_FIELD = "key";
    public static final String VALUE_FIELD = "value";

    public static final String CONNECT_NAME_PROP = "connect.name";
    public static final String CONNECT_DOC_PROP = "connect.doc";
    public static final String CONNECT_VERSION_PROP = "connect.version";
    public static final String CONNECT_DEFAULT_VALUE_PROP = "connect.default";
    public static final String CONNECT_PARAMETERS_PROP = "connect.parameters";
    public static final String CONNECT_INTERNAL_TYPE_NAME = "connect.internal.type";
    public static final String AVRO_RECORD_DOC_PROP = NAMESPACE + ".record.doc";
    public static final String AVRO_ENUM_DOC_PREFIX_PROP = NAMESPACE + ".enum.doc.";
    public static final String AVRO_FIELD_DOC_PREFIX_PROP = NAMESPACE + ".field.doc.";
    //This property is used to determine whether a default value in the Connect schema originated
    //from an Avro field default
    public static final String AVRO_FIELD_DEFAULT_FLAG_PROP = NAMESPACE + ".field.default";
    public static final String AVRO_ENUM_DEFAULT_PREFIX_PROP = NAMESPACE + ".enum.default.";

    public static final String CONNECT_TYPE_PROP = "connect.type";

    public static final String CONNECT_TYPE_INT8 = "int8";
    public static final String CONNECT_TYPE_INT16 = "int16";

    public static final String AVRO_TYPE_UNION = NAMESPACE + ".Union";
    public static final String AVRO_TYPE_ENUM = NAMESPACE + ".Enum";

    public static final String AVRO_TYPE_ANYTHING = NAMESPACE + ".Anything";

    private static final Map<String, Schema.Type> NON_AVRO_TYPES_BY_TYPE_CODE = new HashMap<>();

    static {
        NON_AVRO_TYPES_BY_TYPE_CODE.put(CONNECT_TYPE_INT8, Schema.Type.INT8);
        NON_AVRO_TYPES_BY_TYPE_CODE.put(CONNECT_TYPE_INT16, Schema.Type.INT16);
    }

    // Avro Java object types used by Connect schema types
    private static final Map<Schema.Type, List<Class>> SIMPLE_AVRO_SCHEMA_TYPES = new HashMap<>();

    static {
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.INT32, Arrays.asList((Class) Integer.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.INT64, Arrays.asList((Class) Long.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.FLOAT32, Arrays.asList((Class) Float.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.FLOAT64, Arrays.asList((Class) Double.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.BOOLEAN, Arrays.asList((Class) Boolean.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.STRING, Arrays.asList((Class) CharSequence.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(
                Schema.Type.BYTES,
                Arrays.asList((Class) ByteBuffer.class, (Class) byte[].class, (Class) GenericFixed.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.ARRAY, Arrays.asList((Class) Collection.class));
        SIMPLE_AVRO_SCHEMA_TYPES.put(Schema.Type.MAP, Arrays.asList((Class) Map.class));
    }

    private static final Map<Schema.Type, org.apache.avro.Schema.Type> CONNECT_TYPES_TO_AVRO_TYPES
            = new HashMap<>();

    static {
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.INT32, org.apache.avro.Schema.Type.INT);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.INT64, org.apache.avro.Schema.Type.LONG);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.FLOAT32, org.apache.avro.Schema.Type.FLOAT);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.FLOAT64, org.apache.avro.Schema.Type.DOUBLE);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.BOOLEAN, org.apache.avro.Schema.Type.BOOLEAN);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.STRING, org.apache.avro.Schema.Type.STRING);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.BYTES, org.apache.avro.Schema.Type.BYTES);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.ARRAY, org.apache.avro.Schema.Type.ARRAY);
        CONNECT_TYPES_TO_AVRO_TYPES.put(Schema.Type.MAP, org.apache.avro.Schema.Type.MAP);
    }


    private static final String ANYTHING_SCHEMA_BOOLEAN_FIELD = "boolean";
    private static final String ANYTHING_SCHEMA_BYTES_FIELD = "bytes";
    private static final String ANYTHING_SCHEMA_DOUBLE_FIELD = "double";
    private static final String ANYTHING_SCHEMA_FLOAT_FIELD = "float";
    private static final String ANYTHING_SCHEMA_INT_FIELD = "int";
    private static final String ANYTHING_SCHEMA_LONG_FIELD = "long";
    private static final String ANYTHING_SCHEMA_STRING_FIELD = "string";
    private static final String ANYTHING_SCHEMA_ARRAY_FIELD = "array";
    private static final String ANYTHING_SCHEMA_MAP_FIELD = "map";

    public static final org.apache.avro.Schema ANYTHING_SCHEMA_MAP_ELEMENT;
    public static final org.apache.avro.Schema ANYTHING_SCHEMA;

    private static final org.apache.avro.Schema
            NULL_AVRO_SCHEMA =
            org.apache.avro.Schema.create(org.apache.avro.Schema.Type.NULL);

    static {
        // Intuitively this should be a union schema. However, unions can't be named in Avro and this
        // is a self-referencing type, so we need to use a format in which we can name the entire schema

        ANYTHING_SCHEMA =
                org.apache.avro.SchemaBuilder.record(AVRO_TYPE_ANYTHING).namespace(NAMESPACE).fields()
                        .optionalBoolean(ANYTHING_SCHEMA_BOOLEAN_FIELD)
                        .optionalBytes(ANYTHING_SCHEMA_BYTES_FIELD)
                        .optionalDouble(ANYTHING_SCHEMA_DOUBLE_FIELD)
                        .optionalFloat(ANYTHING_SCHEMA_FLOAT_FIELD)
                        .optionalInt(ANYTHING_SCHEMA_INT_FIELD)
                        .optionalLong(ANYTHING_SCHEMA_LONG_FIELD)
                        .optionalString(ANYTHING_SCHEMA_STRING_FIELD)
                        .name(ANYTHING_SCHEMA_ARRAY_FIELD).type().optional().array()
                        .items().type(AVRO_TYPE_ANYTHING)
                        .name(ANYTHING_SCHEMA_MAP_FIELD).type().optional().array()
                        .items().record(MAP_ENTRY_TYPE_NAME).namespace(NAMESPACE).fields()
                        .name(KEY_FIELD).type(AVRO_TYPE_ANYTHING).noDefault()
                        .name(VALUE_FIELD).type(AVRO_TYPE_ANYTHING).noDefault()
                        .endRecord()
                        .endRecord();
        // This is convenient to have extracted; we can't define it before ANYTHING_SCHEMA because it
        // uses ANYTHING_SCHEMA in its definition.
        ANYTHING_SCHEMA_MAP_ELEMENT = ANYTHING_SCHEMA.getField("map").schema()
                .getTypes().get(1) // The "map" field is optional, get the schema from the union type
                .getElementType();
    }


    // Convert values in Connect form into their logical types. These logical converters are
    // discovered by logical type names specified in the field
    private static final HashMap<String, LogicalTypeConverter> TO_CONNECT_LOGICAL_CONVERTERS
            = new HashMap<>();

    static {
        TO_CONNECT_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (value instanceof byte[]) {
                    return Decimal.toLogical(schema, (byte[]) value);
                } else if (value instanceof ByteBuffer) {
                    return Decimal.toLogical(schema, ((ByteBuffer) value).array());
                }
                throw new DataException(
                        "Invalid type for Decimal, underlying representation should be bytes but was "
                                + value.getClass());
            }
        });

        TO_CONNECT_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof Integer)) {
                    throw new DataException(
                            "Invalid type for Date, underlying representation should be int32 but was "
                                    + value.getClass());
                }
                return Date.toLogical(schema, (int) value);
            }
        });

        TO_CONNECT_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof Integer)) {
                    throw new DataException(
                            "Invalid type for Time, underlying representation should be int32 but was "
                                    + value.getClass());
                }
                return Time.toLogical(schema, (int) value);
            }
        });

        TO_CONNECT_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof Long)) {
                    throw new DataException(
                            "Invalid type for Timestamp, underlying representation should be int64 but was "
                                    + value.getClass());
                }
                return Timestamp.toLogical(schema, (long) value);
            }
        });
    }

    static final String AVRO_PROP = "avro";
    static final String AVRO_LOGICAL_TYPE_PROP = "logicalType";
    static final String AVRO_LOGICAL_TIMESTAMP_MILLIS = "timestamp-millis";
    static final String AVRO_LOGICAL_TIME_MILLIS = "time-millis";
    static final String AVRO_LOGICAL_DATE = "date";
    static final String AVRO_LOGICAL_DECIMAL = "decimal";
    static final String AVRO_LOGICAL_DECIMAL_SCALE_PROP = "scale";
    static final String AVRO_LOGICAL_DECIMAL_PRECISION_PROP = "precision";
    static final String CONNECT_AVRO_DECIMAL_PRECISION_PROP = "connect.decimal.precision";
    static final Integer CONNECT_AVRO_DECIMAL_PRECISION_DEFAULT = 64;

    private static final HashMap<String, LogicalTypeConverter> TO_AVRO_LOGICAL_CONVERTERS
            = new HashMap<>();

    static {
        TO_AVRO_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof BigDecimal)) {
                    throw new DataException(
                            "Invalid type for Decimal, expected BigDecimal but was " + value.getClass());
                }
                return Decimal.fromLogical(schema, (BigDecimal) value);
            }
        });

        TO_AVRO_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof java.util.Date)) {
                    throw new DataException(
                            "Invalid type for Date, expected Date but was " + value.getClass());
                }
                return Date.fromLogical(schema, (java.util.Date) value);
            }
        });

        TO_AVRO_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof java.util.Date)) {
                    throw new DataException(
                            "Invalid type for Time, expected Date but was " + value.getClass());
                }
                return Time.fromLogical(schema, (java.util.Date) value);
            }
        });

        TO_AVRO_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() {
            @Override
            public Object convert(Schema schema, Object value) {
                if (!(value instanceof java.util.Date)) {
                    throw new DataException(
                            "Invalid type for Timestamp, expected Date but was " + value.getClass());
                }
                return Timestamp.fromLogical(schema, (java.util.Date) value);
            }
        });
    }

    private Cache<Schema, org.apache.avro.Schema> fromConnectSchemaCache;
    private Cache<AvroSchema, Schema> toConnectSchemaCache;
    private boolean connectMetaData;
    private boolean enhancedSchemaSupport;

    public AvroData(int cacheSize) {
        this(new AvroDataConfig.Builder()
                .with(AvroDataConfig.SCHEMAS_CACHE_SIZE_CONFIG, cacheSize)
                .build());
    }

    public AvroData(AvroDataConfig avroDataConfig) {
        fromConnectSchemaCache =
                new SynchronizedCache<>(new LRUCache<>(
                        avroDataConfig.getSchemasCacheSize()));
        toConnectSchemaCache =
                new SynchronizedCache<>(new LRUCache<>(
                        avroDataConfig.getSchemasCacheSize()));
        this.connectMetaData = avroDataConfig.isConnectMetaData();
        this.enhancedSchemaSupport = avroDataConfig.isEnhancedAvroSchemaSupport();
    }

    /**
     * Convert this object, in Connect data format, into an Avro object.
     */
    public Object fromConnectData(Schema schema, Object value) {
        org.apache.avro.Schema avroSchema = fromConnectSchema(schema);
        return fromConnectData(schema, avroSchema, value);
    }

    protected Object fromConnectData(Schema schema, org.apache.avro.Schema avroSchema, Object value) {
        return fromConnectData(schema, avroSchema, value, true, false, enhancedSchemaSupport);
    }

    /**
     * Convert from Connect data format to Avro. This version assumes the Avro schema has already
     * been converted and makes the use of NonRecordContainer optional
     *
     * @param schema                         the Connect schema
     * @param avroSchema                     the corresponding
     * @param logicalValue                   the Connect data to convert, which may be a value for
     *                                       a logical type
     * @param requireContainer               if true, wrap primitives, maps, and arrays in a
     *                                       NonRecordContainer before returning them
     * @param requireSchemalessContainerNull if true, use a container representation of null because
     *                                       this is part of struct/array/map and we cannot represent
     *                                       nulls as true null because Anything cannot be a union
     *                                       type; otherwise, this is a top-level value and can return
     *                                       null
     * @return the converted data
     */
    private static Object fromConnectData(
            Schema schema, org.apache.avro.Schema avroSchema,
            Object logicalValue, boolean requireContainer,
            boolean requireSchemalessContainerNull, boolean enhancedSchemaSupport
    ) {
        Schema.Type schemaType = schema != null
                ? schema.type()
                : schemaTypeForSchemalessJavaType(logicalValue);
        if (schemaType == null) {
            // Schemaless null data since schema is null and we got a null schema type from the value
            if (requireSchemalessContainerNull) {
                return new GenericRecordBuilder(ANYTHING_SCHEMA).build();
            } else {
                return null;
            }
        }

        validateSchemaValue(schema, logicalValue);

        if (logicalValue == null) {
            // But if this is schemaless, we may not be able to return null directly
            if (schema == null && requireSchemalessContainerNull) {
                return new GenericRecordBuilder(ANYTHING_SCHEMA).build();
            } else {
                return null;
            }
        }

        // If this is a logical type, convert it from the convenient Java type to the underlying
        // serializeable format
        Object value = logicalValue;
        if (schema != null && schema.name() != null) {
            LogicalTypeConverter logicalConverter = TO_AVRO_LOGICAL_CONVERTERS.get(schema.name());
            if (logicalConverter != null) {
                value = logicalConverter.convert(schema, logicalValue);
            }
        }

        try {
            switch (schemaType) {
                case INT8: {
                    Byte byteValue = (Byte) value; // Check for correct type
                    Integer convertedByteValue = byteValue == null ? null : byteValue.intValue();
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, convertedByteValue, ANYTHING_SCHEMA_INT_FIELD),
                            requireContainer);
                }
                case INT16: {
                    Short shortValue = (Short) value; // Check for correct type
                    Integer convertedShortValue = shortValue == null ? null : shortValue.intValue();
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, convertedShortValue, ANYTHING_SCHEMA_INT_FIELD),
                            requireContainer);
                }

                case INT32:
                    Integer intValue = (Integer) value; // Check for correct type
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_INT_FIELD),
                            requireContainer);
                case INT64:
                    Long longValue = (Long) value; // Check for correct type
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_LONG_FIELD),
                            requireContainer);
                case FLOAT32:
                    Float floatValue = (Float) value; // Check for correct type
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_FLOAT_FIELD),
                            requireContainer);
                case FLOAT64:
                    Double doubleValue = (Double) value; // Check for correct type
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_DOUBLE_FIELD),
                            requireContainer);
                case BOOLEAN:
                    Boolean boolValue = (Boolean) value; // Check for correct type
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_BOOLEAN_FIELD),
                            requireContainer);
                case STRING:
                    if (enhancedSchemaSupport && schema != null && schema.parameters() != null
                            && schema.parameters().containsKey(AVRO_TYPE_ENUM)) {
                        String enumSchemaName = schema.parameters().get(AVRO_TYPE_ENUM);
                        org.apache.avro.Schema enumSchema;
                        if (avroSchema.getType() == org.apache.avro.Schema.Type.UNION) {
                            int enumIndex = avroSchema.getIndexNamed(enumSchemaName);
                            enumSchema = avroSchema.getTypes().get(enumIndex);
                        } else {
                            enumSchema = avroSchema;
                        }
                        value = new GenericData.EnumSymbol(enumSchema, (String) value);
                    } else {
                        String stringValue = (String) value; // Check for correct type
                    }
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, value, ANYTHING_SCHEMA_STRING_FIELD),
                            requireContainer);

                case BYTES: {
                    ByteBuffer bytesValue = value instanceof byte[] ? ByteBuffer.wrap((byte[]) value) :
                            (ByteBuffer) value;
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, bytesValue, ANYTHING_SCHEMA_BYTES_FIELD),
                            requireContainer);
                }

                case ARRAY: {
                    Collection<Object> list = (Collection<Object>) value;
                    List<Object> converted = new ArrayList<>(list.size());
                    Schema elementSchema = schema != null ? schema.valueSchema() : null;
                    org.apache.avro.Schema underlyingAvroSchema = avroSchemaForUnderlyingTypeIfOptional(
                            schema, avroSchema);
                    org.apache.avro.Schema elementAvroSchema =
                            schema != null ? underlyingAvroSchema.getElementType() : ANYTHING_SCHEMA;
                    for (Object val : list) {
                        converted.add(
                                fromConnectData(
                                        elementSchema,
                                        elementAvroSchema,
                                        val,
                                        false,
                                        true,
                                        enhancedSchemaSupport
                                )
                        );
                    }
                    return maybeAddContainer(
                            avroSchema,
                            maybeWrapSchemaless(schema, converted, ANYTHING_SCHEMA_ARRAY_FIELD),
                            requireContainer);
                }

                case MAP: {
                    Map<Object, Object> map = (Map<Object, Object>) value;
                    org.apache.avro.Schema underlyingAvroSchema;
                    if (schema != null && schema.keySchema().type() == Schema.Type.STRING
                            && !schema.keySchema().isOptional()) {

                        underlyingAvroSchema = avroSchemaForUnderlyingTypeIfOptional(schema, avroSchema);
                        Map<String, Object> converted = new HashMap<>();
                        for (Map.Entry<Object, Object> entry : map.entrySet()) {
                            // Key is a String, no conversion needed
                            Object convertedValue = fromConnectData(schema.valueSchema(),
                                    underlyingAvroSchema.getValueType(),
                                    entry.getValue(), false, true, enhancedSchemaSupport
                            );
                            converted.put((String) entry.getKey(), convertedValue);
                        }
                        return maybeAddContainer(avroSchema, converted, requireContainer);
                    } else {
                        List<GenericRecord> converted = new ArrayList<>(map.size());
                        underlyingAvroSchema = avroSchemaForUnderlyingMapEntryType(schema, avroSchema);
                        org.apache.avro.Schema elementSchema =
                                schema != null
                                        ? underlyingAvroSchema.getElementType()
                                        : ANYTHING_SCHEMA_MAP_ELEMENT;
                        org.apache.avro.Schema avroKeySchema = elementSchema.getField(KEY_FIELD).schema();
                        org.apache.avro.Schema avroValueSchema = elementSchema.getField(VALUE_FIELD).schema();
                        for (Map.Entry<Object, Object> entry : map.entrySet()) {
                            Object keyConverted = fromConnectData(schema != null ? schema.keySchema() : null,
                                    avroKeySchema, entry.getKey(), false, true,
                                    enhancedSchemaSupport);
                            Object valueConverted = fromConnectData(schema != null ? schema.valueSchema() : null,
                                    avroValueSchema, entry.getValue(), false,
                                    true, enhancedSchemaSupport);
                            converted.add(
                                    new GenericRecordBuilder(elementSchema)
                                            .set(KEY_FIELD, keyConverted)
                                            .set(VALUE_FIELD, valueConverted)
                                            .build()
                            );
                        }
                        return maybeAddContainer(
                                avroSchema, maybeWrapSchemaless(schema, converted, ANYTHING_SCHEMA_MAP_FIELD),
                                requireContainer);
                    }
                }

                case STRUCT: {
                    Struct struct = (Struct) value;
                    if (!struct.schema().equals(schema)) {
                        throw new DataException("Mismatching struct schema");
                    }
                    //This handles the inverting of a union which is held as a struct, where each field is
                    // one of the union types.
                    if (AVRO_TYPE_UNION.equals(schema.name())) {
                        for (Field field : schema.fields()) {
                            Object object = struct.get(field);
                            if (object != null) {
                                return fromConnectData(
                                        field.schema(),
                                        avroSchema,
                                        object,
                                        false,
                                        true,
                                        enhancedSchemaSupport
                                );
                            }
                        }
                        return fromConnectData(schema, avroSchema, null, false, true, enhancedSchemaSupport);
                    } else {
                        org.apache.avro.Schema underlyingAvroSchema = avroSchemaForUnderlyingTypeIfOptional(
                                schema, avroSchema);
                        GenericRecordBuilder convertedBuilder = new GenericRecordBuilder(underlyingAvroSchema);
                        for (Field field : schema.fields()) {
                            org.apache.avro.Schema.Field theField = underlyingAvroSchema.getField(field.name());
                            org.apache.avro.Schema fieldAvroSchema = theField.schema();
                            convertedBuilder.set(
                                    field.name(),
                                    fromConnectData(field.schema(), fieldAvroSchema, struct.get(field), false,
                                            true, enhancedSchemaSupport)
                            );
                        }
                        return convertedBuilder.build();
                    }
                }

                default:
                    throw new DataException("Unknown schema type: " + schema.type());
            }
        } catch (ClassCastException e) {
            throw new DataException("Invalid type for " + schema.type() + ": " + value.getClass());
        }
    }

    /**
     * MapEntry types in connect Schemas are represented as Arrays of record.
     * Return the array type from the union instead of the union itself.
     */
    private static org.apache.avro.Schema avroSchemaForUnderlyingMapEntryType(
            Schema schema,
            org.apache.avro.Schema avroSchema) {

        if (schema != null && schema.isOptional()) {
            if (avroSchema.getType() == org.apache.avro.Schema.Type.UNION) {
                for (org.apache.avro.Schema typeSchema : avroSchema.getTypes()) {
                    if (!typeSchema.getType().equals(org.apache.avro.Schema.Type.NULL)
                            && Schema.Type.ARRAY.getName().equals(typeSchema.getType().getName())) {
                        return typeSchema;
                    }
                }
            } else {
                throw new DataException(
                        "An optional schema should have an Avro Union type, not "
                                + schema.type());
            }
        }
        return avroSchema;
    }

    private static boolean crossReferenceSchemaNames(final Schema schema,
                                                     final org.apache.avro.Schema avroSchema) {
        return Objects.equals(avroSchema.getFullName(), schema.name())
                || Objects.equals(avroSchema.getType().getName(), schema.type().getName())
                || (schema.name() == null && avroSchema.getFullName().equals(DEFAULT_SCHEMA_FULL_NAME));
    }

    /**
     * Connect optional fields are represented as a unions (null & type) in Avro
     * Return the Avro schema of the actual type in the Union (instead of the union itself)
     */
    private static org.apache.avro.Schema avroSchemaForUnderlyingTypeIfOptional(
            Schema schema, org.apache.avro.Schema avroSchema) {

        if (schema != null && schema.isOptional()) {
            if (avroSchema.getType() == org.apache.avro.Schema.Type.UNION) {
                for (org.apache.avro.Schema typeSchema : avroSchema
