/*
 * 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.connect.avro.EnumUnion;
import com.connect.avro.UserType;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.ImmutableMap;
import org.apache.avro.LogicalTypes;
import org.apache.avro.generic.GenericContainer;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.avro.util.Utf8;
import org.apache.kafka.common.cache.Cache;
import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.Decimal;
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.hamcrest.core.IsEqual;
import org.junit.Assert;
import org.junit.Test;
import org.powermock.reflect.Whitebox;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
import java.util.stream.Collectors;

import foo.bar.EnumTest;
import foo.bar.Kind;

import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.AVRO_ENUM_DOC_PREFIX_PROP;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.AVRO_FIELD_DEFAULT_FLAG_PROP;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.AVRO_FIELD_DOC_PREFIX_PROP;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.AVRO_TYPE_ENUM;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.CONNECT_INTERNAL_TYPE_NAME;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.CONNECT_NAME_PROP;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.KEY_FIELD;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.MAP_ENTRY_TYPE_NAME;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.NAMESPACE;
import static com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.AvroData.VALUE_FIELD;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertArrayEquals;

public class AvroDataTest {
    private static final int TEST_SCALE = 2;
    private static final BigDecimal TEST_DECIMAL = new BigDecimal(new BigInteger("156"), TEST_SCALE);
    private static final byte[] TEST_DECIMAL_BYTES = new byte[]{0, -100};

    private static final GregorianCalendar EPOCH;
    private static final GregorianCalendar EPOCH_PLUS_TEN_THOUSAND_DAYS;
    private static final GregorianCalendar EPOCH_PLUS_TEN_THOUSAND_MILLIS;
    static {
        EPOCH = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
        EPOCH.setTimeZone(TimeZone.getTimeZone("UTC"));

        EPOCH_PLUS_TEN_THOUSAND_DAYS = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
        EPOCH_PLUS_TEN_THOUSAND_DAYS.setTimeZone(TimeZone.getTimeZone("UTC"));
        EPOCH_PLUS_TEN_THOUSAND_DAYS.add(Calendar.DATE, 10000);

        EPOCH_PLUS_TEN_THOUSAND_MILLIS = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
        EPOCH_PLUS_TEN_THOUSAND_MILLIS.setTimeZone(TimeZone.getTimeZone("UTC"));
        EPOCH_PLUS_TEN_THOUSAND_MILLIS.add(Calendar.MILLISECOND, 10000);
    }

    private static final Schema NAMED_MAP_SCHEMA =
            SchemaBuilder.map(Schema.OPTIONAL_STRING_SCHEMA, Schema.INT32_SCHEMA)
                    .name("foo.bar")
                    .build();
    private static final org.apache.avro.Schema NAMED_AVRO_MAP_SCHEMA =
            org.apache.avro.SchemaBuilder.array()
                    .prop(CONNECT_NAME_PROP, "foo.bar")
                    .items(
                            org.apache.avro.SchemaBuilder.record("foo.bar")
                                    .prop(CONNECT_INTERNAL_TYPE_NAME, MAP_ENTRY_TYPE_NAME)
                                    .fields()
                                    .optionalString(KEY_FIELD)
                                    .requiredInt(VALUE_FIELD)
                                    .endRecord()
                    );

    private AvroData avroData = new AvroData(2);

    // Connect -> Avro

    @Test
    public void testFromConnectBoolean() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().booleanType();
        checkNonRecordConversion(avroSchema, true, Schema.BOOLEAN_SCHEMA, true, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_BOOLEAN_SCHEMA);
    }

    @Test
    public void testFromConnectByte() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType();
        avroSchema.addProp("connect.type", "int8");
        checkNonRecordConversion(avroSchema, 12, Schema.INT8_SCHEMA, (byte) 12, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_INT8_SCHEMA);
    }

    @Test
    public void testFromConnectShort() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType();
        avroSchema.addProp("connect.type", "int16");
        checkNonRecordConversion(avroSchema, 12, Schema.INT16_SCHEMA, (short) 12, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_INT16_SCHEMA);
    }

    @Test
    public void testFromConnectInteger() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType();
        checkNonRecordConversion(avroSchema, 12, Schema.INT32_SCHEMA, 12, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_INT32_SCHEMA);
    }

    @Test
    public void testFromConnectLong() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().longType();
        checkNonRecordConversion(avroSchema, 12L, Schema.INT64_SCHEMA, 12L, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_INT64_SCHEMA);
    }

    @Test
    public void testFromConnectFloat() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().floatType();
        checkNonRecordConversion(avroSchema, 12.2f, Schema.FLOAT32_SCHEMA, 12.2f, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_FLOAT32_SCHEMA);
    }

    @Test
    public void testFromConnectDouble() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().doubleType();
        checkNonRecordConversion(avroSchema, 12.2, Schema.FLOAT64_SCHEMA, 12.2, avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_FLOAT64_SCHEMA);
    }

    @Test
    public void testFromConnectBytes() {
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().bytesType();
        checkNonRecordConversion(avroSchema, ByteBuffer.wrap("foo".getBytes()),
                Schema.BYTES_SCHEMA, "foo".getBytes(), avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_BYTES_SCHEMA);
    }

    @Test
    public void testFromConnectString() {
        org.apache.avro.Schema avroSchema =
                org.apache.avro.SchemaBuilder.builder().stringType();
        checkNonRecordConversion(avroSchema, "string", Schema.STRING_SCHEMA, "string", avroData);

        checkNonRecordConversionNull(Schema.OPTIONAL_STRING_SCHEMA);
    }

    @Test
    public void testFromConnectEnum() {
        AvroDataConfig avroDataConfig = new AvroDataConfig.Builder()
                .with(AvroDataConfig.ENHANCED_AVRO_SCHEMA_SUPPORT_CONFIG, true)
                .build();
        AvroData avroData = new AvroData(avroDataConfig);

        org.apache.avro.Schema avroSchema =
                org.apache.avro.SchemaBuilder.builder().enumeration("enum").symbols("one","two","three");
        GenericData.EnumSymbol avroObj = new GenericData.EnumSymbol(avroSchema, "one");

        Map connectPropsMap = ImmutableMap.of("connect.enum.doc","null",
                "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.Enum","enum",
                "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.Enum.one", "one",
                "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.Enum.two","two",
                "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata.Enum.three","three");
        avroSchema.addProp("connect.parameters", connectPropsMap);
        avroSchema.addProp("connect.name", "enum");
        SchemaAndValue schemaAndValue = avroData.toConnectData(avroSchema, avroObj);
        checkNonRecordConversion(avroSchema, avroObj, schemaAndValue.schema(), schemaAndValue.value(),
                avroData);
    }

    @Test
    public void testFromConnectMapWithStringKey() {
        final Schema schema = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA);
        final org.apache.avro.Schema expected = org.apache.avro.SchemaBuilder.map()
                .values(org.apache.avro.SchemaBuilder.builder().intType());
        assertThat(avroData.fromConnectSchema(schema), equalTo(expected));
    }

    @Test
    public void testFromConnectMapWithOptionalKey() {
        final Schema schema = SchemaBuilder.map(Schema.OPTIONAL_STRING_SCHEMA, Schema.INT32_SCHEMA);
        final org.apache.avro.Schema expected = org.apache.avro.SchemaBuilder.array()
                .items(
                        org.apache.avro.SchemaBuilder.record(NAMESPACE + "." + MAP_ENTRY_TYPE_NAME)
                                .fields()
                                .optionalString(KEY_FIELD)
                                .requiredInt(VALUE_FIELD)
                                .endRecord()
                );

        assertThat(avroData.fromConnectSchema(schema), equalTo(expected));
    }

    @Test
    public void testFromConnectMapWithNonStringKey() {
        final Schema schema = SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.INT32_SCHEMA);
        final org.apache.avro.Schema expected = org.apache.avro.SchemaBuilder.array()
                .items(
                        org.apache.avro.SchemaBuilder.record(NAMESPACE + "." + MAP_ENTRY_TYPE_NAME)
                                .fields()
                                .requiredInt(KEY_FIELD)
                                .requiredInt(VALUE_FIELD)
                                .endRecord()
                );

        assertThat(avroData.fromConnectSchema(schema), equalTo(expected));
    }

    @Test
    public void testFromNamedConnectMapWithNonStringKey() {
        assertThat(avroData.fromConnectSchema(NAMED_MAP_SCHEMA), equalTo(NAMED_AVRO_MAP_SCHEMA));
    }

    @Test
    public void testFromConnectComplex() {
        Schema schema = SchemaBuilder.struct()
                .field("int8", SchemaBuilder.int8().defaultValue((byte) 2).doc("int8 field").build())
                .field("int16", Schema.INT16_SCHEMA)
                .field("int32", Schema.INT32_SCHEMA)
                .field("int64", Schema.INT64_SCHEMA)
                .field("float32", Schema.FLOAT32_SCHEMA)
                .field("float64", Schema.FLOAT64_SCHEMA)
                .field("boolean", Schema.BOOLEAN_SCHEMA)
                .field("string", Schema.STRING_SCHEMA)
                .field("bytes", Schema.BYTES_SCHEMA)
                .field("array", SchemaBuilder.array(Schema.STRING_SCHEMA).build())
                .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build())
                .field("mapNonStringKeys",
                        SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.INT32_SCHEMA).build())
                .build();
        Struct struct = new Struct(schema)
                .put("int8", (byte) 12)
                .put("int16", (short) 12)
                .put("int32", 12)
                .put("int64", 12L)
                .put("float32", 12.2f)
                .put("float64", 12.2)
                .put("boolean", true)
                .put("string", "foo")
                .put("bytes", ByteBuffer.wrap("foo".getBytes()))
                .put("array", Arrays.asList("a", "b", "c"))
                .put("map", Collections.singletonMap("field", 1))
                .put("mapNonStringKeys", Collections.singletonMap(1, 1));

        Object convertedRecord = avroData.fromConnectData(schema, struct);

        org.apache.avro.Schema complexMapElementSchema =
                org.apache.avro.SchemaBuilder
                        .record("MapEntry").namespace("com.amazonaws.services.schemaregistry.kafkaconnect.avrodata").fields()
                        .requiredInt("key")
                        .requiredInt("value")
                        .endRecord();

        // One field has some extra data set on it to ensure it gets passed through via the fields
        // config
        org.apache.avro.Schema int8Schema = org.apache.avro.SchemaBuilder.builder().intType();
        int8Schema.addProp("connect.doc", "int8 field");
        int8Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(2));
        int8Schema.addProp("connect.type", "int8");
        org.apache.avro.Schema int16Schema = org.apache.avro.SchemaBuilder.builder().intType();
        int16Schema.addProp("connect.type", "int16");
        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder
                .record(AvroData.DEFAULT_SCHEMA_NAME).namespace(AvroData.NAMESPACE) // default values
                .fields()
                .name("int8").type(int8Schema).withDefault(2)
                .name("int16").type(int16Schema).noDefault()
                .requiredInt("int32")
                .requiredLong("int64")
                .requiredFloat("float32")
                .requiredDouble("float64")
                .requiredBoolean("boolean")
                .requiredString("string")
                .requiredBytes("bytes")
                .name("array").type().array().items().stringType().noDefault()
                .name("map").type().map().values().intType().noDefault()
                .name("mapNonStringKeys").type().array().items(complexMapElementSchema).noDefault()
                .endRecord();
        org.apache.avro.generic.GenericRecord avroRecord
                = new org.apache.avro.generic.GenericRecordBuilder(avroSchema)
                .set("int8", 12)
                .set("int16", 12)
                .set("int32", 12)
                .set("int64", 12L)
                .set("float32", 12.2f)
                .set("float64", 12.2)
                .set("boolean", true)
                .set("string", "foo")
                .set("bytes", ByteBuffer.wrap("foo".getBytes()))
                .set("array", Arrays.asList("a", "b", "c"))
                .set("map", Collections.singletonMap("field", 1))
                .set("mapNonStringKeys", Arrays.asList(
                        new GenericRecordBuilder(complexMapElementSchema)
                                .set(AvroData.KEY_FIELD, 1)
                                .set(AvroData.VALUE_FIELD, 1)
                                .build()))
                .build();

        assertEquals(avroSchema, ((org.apache.avro.generic.GenericRecord) convertedRecord).getSchema());
        assertEquals(avroRecord, convertedRecord);
    }

    @Test
    public void testFromConnectComplexWithDefaults() {
        int dateDefVal = 100;
        int timeDefVal = 1000 * 60 * 60 * 2;
        long tsDefVal = 1000 * 60 * 60 * 24 * 365 + 100;
        java.util.Date dateDef = Date.toLogical(Date.SCHEMA, dateDefVal);
        java.util.Date timeDef = Time.toLogical(Time.SCHEMA, timeDefVal);
        java.util.Date tsDef = Timestamp.toLogical(Timestamp.SCHEMA, tsDefVal);
        BigDecimal decimalDef = new BigDecimal(BigInteger.valueOf(314159L), 5);
        byte[] decimalDefVal = decimalDef.unscaledValue().toByteArray();

        Schema schema = SchemaBuilder.struct()
                .field("int8", SchemaBuilder.int8().defaultValue((byte) 2).doc("int8 field").build())
                .field("int16", SchemaBuilder.int16().defaultValue((short)12).doc("int16 field").build())
                .field("int32", SchemaBuilder.int32().defaultValue(12).doc("int32 field").build())
                .field("int64", SchemaBuilder.int64().defaultValue(12L).doc("int64 field").build())
                .field("float32", SchemaBuilder.float32().defaultValue(12.2f).doc("float32 field").build())
                .field("float64", SchemaBuilder.float64().defaultValue(12.2).doc("float64 field").build())
                .field("boolean", SchemaBuilder.bool().defaultValue(true).doc("bool field").build())
                .field("string", SchemaBuilder.string().defaultValue("foo").doc("string field").build())
                .field("bytes", SchemaBuilder.bytes().defaultValue(ByteBuffer.wrap("foo".getBytes())).doc("bytes field").build())
                .field("array", SchemaBuilder.array(Schema.STRING_SCHEMA).defaultValue(Arrays.asList("a", "b", "c")).build())
                .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).defaultValue(Collections.singletonMap("field", 1)).build())
                .field("date", Date.builder().defaultValue(dateDef).doc("date field").build())
                .field("time", Time.builder().defaultValue(timeDef).doc("time field").build())
                .field("ts", Timestamp.builder().defaultValue(tsDef).doc("ts field").build())
                .field("decimal", Decimal.builder(5).defaultValue(decimalDef).doc("decimal field").build())
                .build();
        // leave the struct empty so that only defaults are used
        Struct struct = new Struct(schema)
                .put("int8", (byte) 2)
                .put("int16", (short) 12)
                .put("int32", 12)
                .put("int64", 12L)
                .put("float32", 12.2f)
                .put("float64", 12.2)
                .put("boolean", true)
                .put("string", "foo")
                .put("bytes", ByteBuffer.wrap("foo".getBytes()))
                .put("array", Arrays.asList("a", "b", "c"))
                .put("map", Collections.singletonMap("field", 1))
                .put("date", dateDef)
                .put("time", timeDef)
                .put("ts", tsDef)
                .put("decimal", decimalDef);

        // Define the expected Avro schema
        org.apache.avro.Schema complexMapElementSchema =
                org.apache.avro.SchemaBuilder
                        .record("MapEntry").namespace("com.amazonaws.services.schemaregistry.kafkaconnect.avrodata").fields()
                        .requiredInt("key")
                        .requiredInt("value")
                        .endRecord();

        org.apache.avro.Schema int8Schema = org.apache.avro.SchemaBuilder.builder().intType();
        int8Schema.addProp("connect.doc", "int8 field");
        int8Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(2));
        int8Schema.addProp("connect.type", "int8");
        org.apache.avro.Schema int16Schema = org.apache.avro.SchemaBuilder.builder().intType();
        int16Schema.addProp("connect.doc", "int16 field");
        int16Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(((short)12)).intValue());
        int16Schema.addProp("connect.type", "int16");
        org.apache.avro.Schema int32Schema = org.apache.avro.SchemaBuilder.builder().intType();
        int32Schema.addProp("connect.doc", "int32 field");
        int32Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(12));
        org.apache.avro.Schema int64Schema = org.apache.avro.SchemaBuilder.builder().longType();
        int64Schema.addProp("connect.doc", "int64 field");
        int64Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(12L));
        org.apache.avro.Schema float32Schema = org.apache.avro.SchemaBuilder.builder().floatType();
        float32Schema.addProp("connect.doc", "float32 field");
        float32Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(12.2f));
        org.apache.avro.Schema float64Schema = org.apache.avro.SchemaBuilder.builder().doubleType();
        float64Schema.addProp("connect.doc", "float64 field");
        float64Schema.addProp("connect.default", JsonNodeFactory.instance.numberNode(12.2));
        org.apache.avro.Schema boolSchema = org.apache.avro.SchemaBuilder.builder().booleanType();
        boolSchema.addProp("connect.doc", "bool field");
        boolSchema.addProp("connect.default", JsonNodeFactory.instance.booleanNode(true));
        org.apache.avro.Schema stringSchema = org.apache.avro.SchemaBuilder.builder().stringType();
        stringSchema.addProp("connect.doc", "string field");
        stringSchema.addProp("connect.default", JsonNodeFactory.instance.textNode("foo"));
        org.apache.avro.Schema bytesSchema = org.apache.avro.SchemaBuilder.builder().bytesType();
        bytesSchema.addProp("connect.doc", "bytes field");
        bytesSchema.addProp("connect.default", JsonNodeFactory.instance.textNode(
                new String("foo".getBytes(), StandardCharsets.ISO_8859_1)
        ));

        org.apache.avro.Schema dateSchema = org.apache.avro.SchemaBuilder.builder().intType();
        dateSchema.addProp("connect.doc", "date field");
        dateSchema.addProp("connect.default", JsonNodeFactory.instance.numberNode(dateDefVal));
        dateSchema.addProp(AvroData.CONNECT_NAME_PROP, Date.LOGICAL_NAME);
        dateSchema.addProp(AvroData.CONNECT_VERSION_PROP, 1);
        // this is the new and correct way to set logical type
        LogicalTypes.date().addToSchema(dateSchema);
        // this is the old and wrong way to set logical type
        // leave the line here for back compatibility
        dateSchema.addProp(AvroData.AVRO_LOGICAL_TYPE_PROP, AvroData.AVRO_LOGICAL_DATE);

        org.apache.avro.Schema timeSchema = org.apache.avro.SchemaBuilder.builder().intType();
        timeSchema.addProp("connect.doc", "time field");
        timeSchema.addProp("connect.default", JsonNodeFactory.instance.numberNode(timeDefVal));
        timeSchema.addProp(AvroData.CONNECT_NAME_PROP, Time.LOGICAL_NAME);
        timeSchema.addProp(AvroData.CONNECT_VERSION_PROP, 1);
        // this is the new and correct way to set logical type
        LogicalTypes.timeMillis().addToSchema(timeSchema);
        // this is the old and wrong way to set logical type
        // leave the line here for back compatibility
        timeSchema.addProp(AvroData.AVRO_LOGICAL_TYPE_PROP, AvroData.AVRO_LOGICAL_TIME_MILLIS);

        org.apache.avro.Schema tsSchema = org.apache.avro.SchemaBuilder.builder().longType();
        tsSchema.addProp("connect.doc", "ts field");
        tsSchema.addProp("connect.default", JsonNodeFactory.instance.numberNode(tsDefVal));
        tsSchema.addProp(AvroData.CONNECT_NAME_PROP, Timestamp.LOGICAL_NAME);
        tsSchema.addProp(AvroData.CONNECT_VERSION_PROP, 1);
        // this is the new and correct way to set logical type
        LogicalTypes.timestampMillis().addToSchema(tsSchema);
        // this is the old and wrong way to set logical type
        // leave the line here for back compatibility
        tsSchema.addProp(AvroData.AVRO_LOGICAL_TYPE_PROP, AvroData.AVRO_LOGICAL_TIMESTAMP_MILLIS);

        org.apache.avro.Schema decimalSchema = org.apache.avro.SchemaBuilder.builder().bytesType();
        decimalSchema.addProp("scale", 5);
        decimalSchema.addProp("precision", 64);
        decimalSchema.addProp("connect.doc", "decimal field");
        decimalSchema.addProp(AvroData.CONNECT_VERSION_PROP, 1);
        decimalSchema.addProp("connect.default", JsonNodeFactory.instance.textNode(
                new String(decimalDefVal, StandardCharsets.ISO_8859_1)));
        decimalSchema.addProp("connect.parameters", parameters("scale", "5"));
        decimalSchema.addProp(AvroData.CONNECT_NAME_PROP, Decimal.LOGICAL_NAME);
        // this is the new and correct way to set logical type
        LogicalTypes.decimal(64, 5).addToSchema(decimalSchema);
        // this is the old and wrong way to set logical type
        // leave the line here for back compatibility
        decimalSchema.addProp(AvroData.AVRO_LOGICAL_TYPE_PROP, AvroData.AVRO_LOGICAL_DECIMAL);

        org.apache.avro.Schema arraySchema = org.apache.avro.SchemaBuilder.builder().array().items().stringType();
        ArrayNode arrayNode = JsonNodeFactory.instance.arrayNode();
        arrayNode.add("a");
        arrayNode.add("b");
        arrayNode.add("c");
        arraySchema.addProp("connect.default", arrayNode);

        org.apache.avro.Schema mapSchema = org.apache.avro.SchemaBuilder.builder().map().values().intType();
        ObjectNode mapNode = JsonNodeFactory.instance.objectNode();
        mapNode.put("field", 1);
        mapSchema.addProp("connect.default", mapNode);

        org.apache.avro.Schema nonStringMapSchema = org.apache.avro.SchemaBuilder.builder().array()
                .items(complexMapElementSchema);
        ArrayNode nonStringMapNode = JsonNodeFactory.instance.arrayNode();
        nonStringMapNode.add(JsonNodeFactory.instance.numberNode(1));
        nonStringMapNode.add(JsonNodeFactory.instance.numberNode(1));
        ArrayNode nonStringMapArrayNode = JsonNodeFactory.instance.arrayNode();
        nonStringMapArrayNode.add(nonStringMapNode);
        nonStringMapSchema.addProp("connect.default", nonStringMapArrayNode);

        org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder
                .record(AvroData.DEFAULT_SCHEMA_NAME).namespace(AvroData.NAMESPACE) // default values
                .fields()
                .name("int8").type(int8Schema).withDefault(2)
                .name("int16").type(int16Schema).withDefault(12)
                .name("int32").type(int32Schema).withDefault(12)
                .name("int64").type(int64Schema).withDefault(12L)
                .name("float32").type(float32Schema).withDefault(12.2f)
                .name("float64").type(float64Schema).withDefault(12.2)
                .name("boolean").type(boolSchema).withDefault(true)
                .name("string").type(stringSchema).withDefault("foo")
                .name("bytes").type(bytesSchema).withDefault(ByteBuffer.wrap("foo".getBytes()))
                .name("array").type(arraySchema).withDefault(Arrays.asList("a", "b", "c"))
                .name("map").type(mapSchema).withDefault(Collections.singletonMap("field", 1))
                .name("date").type(dateSchema).withDefault(dateDefVal)
                .name("time").type(timeSchema).withDefault(timeDefVal)
                .name("ts").type(tsSchema).withDefault(tsDefVal)
                .name("decimal").type(decimalSchema).withDefault(ByteBuffer.wrap(decimalDefVal))
                .endRecord();
        org.apache.avro.generic.GenericRecord avroRecord
                = new org.apache.avro.generic.GenericRecordBuilder(avroSchema)
                .set("int8", 2)
                .set("int16", 12)
                .set("int32", 12)
                .set("int64", 12L)
                .set("float32", 12.2f)
                .set("float64", 12.2)
                .set("boolean", true)
                .set("string", "foo")
                .set("bytes", ByteBuffer.wrap("foo".getBytes()))
                .set("array", Arrays.asList("a", "b", "c"))
                .set("map", Collections.singletonMap("field", 1))
                .set("date", dateDefVal)
                .set("time", timeDefVal)
                .set("ts", tsDefVal)
                .set("decimal", decimalDefVal)
                .build();

        SchemaAndValue schemaAndValue = new SchemaAndValue(schema, struct);
        schemaAndValue = convertToConnect(avroSchema, avroRecord, schemaAndValue);
        schemaAndValue = convertToConnect(avroSchema, avroRecord, schemaAndValue);
        schemaAndValue = convertToConnect(avroSchema, avroRecord, schemaAndValue);
        assertNotNull(schemaAndValue);
    }

    protected SchemaAndValue convertToConnect(
            org.apache.avro.Schema expectedAvroSchema,
            org.apache.avro.generic.GenericRecord expectedAvroRecord,
            SchemaAndValue connectSchemaAndValue
    ) {
        Object convertedRecord = avroData.fromConnectData(
                connectSchemaAndValue.schema(),
                connectSchemaAndValue.value()
        );

        org.apache.avro.generic.GenericRecord convertedAvroRecord =
                (org.apache.avro.generic.GenericRecord) convertedRecord;
        assertSchemaEquals(expectedAvroSchema, convertedAvroRecord.getSchema());
        assertSchemaEquals(expectedAvroRecord.getSchema(), convertedAvroRecord.getSchema());

        // This doesn't work because the long field's default value is an integer
        //assertEquals(avroRecord, convertedRecord);
        // We've already checked the schemas, so we just need to check the record field values
        for (org.apache.avro.Schema.Field field : expectedAvroSchema.getFields()) {
            Object actual = convertedAvroRecord.get(field.name());
            Object expected = expectedAvroRecord.get(field.name());
            assertValueEquals(expected, actual);
        }

        SchemaAndValue schemaAndValue = avroData.toConnectData(
                convertedAvroRecord.getSchema(),
                convertedRecord
        );
        assertEquals(connectSchemaAndValue.schema(), schemaAndValue.schema());
        assertEquals(connectSchemaAndValue.value(), schemaAndValue.value());
        return schemaAndValue;
    }

    @Test
    public void testFromConnectOptionalWithDefaultNull() {
        Schema schema = SchemaBuilder.struct()
                .field("optionalBool", SchemaBuilder.bool().optional().defaultValue(null).build())
                .build();
        org.apache.avro.Schema avroSchema = avroData.fromConnectSchema(schema);
        org.apache.avro.Schema expectedAvroSchema = org.apache.avro.SchemaBuilder.builder()
                .record("ConnectDefault").namespace("com.amazonaws.services.schemaregistry.kafkaconnect.avrodata").fields()
                .name("optionalBool").type(org.apache.avro.SchemaBuilder.builder()
                        .unionOf().nullType().and().booleanType().endUnion()).withDefault(null)
                .endRecord();

        assertEquals(expectedAvroSchema, avroSchema);

        Struct struct = new Struct(schema)
                .put("optionalBool", true);
        Object convertedRecord = avroData.fromConnectData(schema, struct);
        org.apache.avro.generic.GenericRecord avroRecord = new org.apache.avro.generic.GenericRecordBuilder(avroSchema)
                .set("optionalBool", true)
                .build();

        assertEquals(avroRecord, convertedRecord);
    }

    @Test
    public void testFromConnectOptionalAnonymousStruct() {
        Schema schema = SchemaBuilder.struct().optional()
                .field("int32", Schema.INT32_SCHEMA)
                .build();

        Struct struct = new Struct(schema).put("int32", 12);

        Object convertedRecord = avroData.fromConnectData(schema, struct);

        assertThat(convertedRecord, instanceOf(GenericRecord.class));
        assertThat(((GenericRecord)convertedRecord).get("int32"), equalTo(12));
    }

    @Test
    public void testFromConnectOptionalComplex() {
