Base Java S3: Update Avro TimeWithTimezone schema mapping (#21909)

* 21908 Base Java S3: Update Avro TimeWithTimezone schema mapping

* 21908 Base Java S3: Formatting

* 21908 Base Java S3: fix integration test gcs + S3

* 21908 Base Java S3: fix unit test

* 21908 Base Java S3: fix format

Author: suhomud

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/JsonSchemaType.java

@@ -23,7 +23,7 @@ public enum JsonSchemaType {
   DATE_V1("WellKnownTypes.json#/definitions/Date", Schema.Type.INT),
   TIMESTAMP_WITH_TIMEZONE_V1("WellKnownTypes.json#/definitions/TimestampWithTimezone", Schema.Type.LONG),
   TIMESTAMP_WITHOUT_TIMEZONE_V1("WellKnownTypes.json#/definitions/TimestampWithoutTimezone", Schema.Type.LONG),
-  TIME_WITH_TIMEZONE_V1("WellKnownTypes.json#/definitions/TimeWithTimezone", Schema.Type.LONG),
+  TIME_WITH_TIMEZONE_V1("WellKnownTypes.json#/definitions/TimeWithTimezone", Schema.Type.STRING),
   TIME_WITHOUT_TIMEZONE_V1("WellKnownTypes.json#/definitions/TimeWithoutTimezone", Schema.Type.LONG),
   OBJECT("object", Schema.Type.RECORD),
   ARRAY("array", Schema.Type.ARRAY),

airbyte-integrations/bases/base-java-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/JsonToAvroSchemaConverter.java

@@ -225,11 +225,12 @@ Schema parseSingleType(final String fieldName,
 
     final Schema fieldSchema;
     switch (fieldType) {
-      case INTEGER_V1, NUMBER_V1, BOOLEAN_V1, STRING_V1, BINARY_DATA_V1 -> fieldSchema = Schema.create(fieldType.getAvroType());
+      case INTEGER_V1, NUMBER_V1, BOOLEAN_V1, STRING_V1, TIME_WITH_TIMEZONE_V1, BINARY_DATA_V1 -> fieldSchema =
+          Schema.create(fieldType.getAvroType());
       case DATE_V1 -> fieldSchema = LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT));
       case TIMESTAMP_WITH_TIMEZONE_V1, TIMESTAMP_WITHOUT_TIMEZONE_V1 -> fieldSchema = LogicalTypes.timestampMicros()
           .addToSchema(Schema.create(Schema.Type.LONG));
-      case TIME_WITH_TIMEZONE_V1, TIME_WITHOUT_TIMEZONE_V1 -> fieldSchema = LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG));
+      case TIME_WITHOUT_TIMEZONE_V1 -> fieldSchema = LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG));
       case INTEGER_V0, NUMBER_V0, NUMBER_INT_V0, NUMBER_BIGINT_V0, NUMBER_FLOAT_V0, BOOLEAN_V0 -> fieldSchema =
           Schema.create(fieldType.getAvroType());
       case STRING_V0 -> {

airbyte-integrations/bases/base-java-s3/src/test/resources/parquet/json_schema_converter/type_conversion_test_cases_v1.json

@@ -204,11 +204,7 @@
     "jsonFieldSchema": {
       "$ref": "WellKnownTypes.json#/definitions/TimeWithTimezone"
     },
-    "avroFieldType": [
-      "null",
-      { "type": "long", "logicalType": "time-micros" },
-      "string"
-    ]
+    "avroFieldType": ["null", "string"]
   },
   {
     "fieldName": "array_field_without_items",

airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/comparator/AdvancedTestDataComparator.java

@@ -105,8 +105,10 @@ protected boolean compareJsonNodes(final JsonNode expectedValue, final JsonNode
       return compareDateTimeValues(expectedValue.asText(), actualValue.asText());
     } else if (isDateValue(expectedValue.asText())) {
       return compareDateValues(expectedValue.asText(), actualValue.asText());
-    } else if (isTimeWithTimezone(expectedValue.asText()) || isTimeWithoutTimezone(expectedValue.asText())) {
-      return compareTime(expectedValue.asText(), actualValue.asText());
+    } else if (isTimeWithTimezone(expectedValue.asText())) {
+      return compareTimeWithTimeZone(expectedValue.asText(), actualValue.asText());
+    } else if (isTimeWithoutTimezone(expectedValue.asText())) {
+      return compareTimeWithoutTimeZone(expectedValue.asText(), actualValue.asText());
     } else if (expectedValue.isArray()) {
       return compareArrays(expectedValue, actualValue);
     } else if (expectedValue.isObject()) {
@@ -215,7 +217,11 @@ protected boolean compareDateValues(final String airbyteMessageValue, final Stri
     return compareTextValues(airbyteMessageValue, destinationValue);
   }
 
-  protected boolean compareTime(final String airbyteMessageValue, final String destinationValue) {
+  protected boolean compareTimeWithoutTimeZone(final String airbyteMessageValue, final String destinationValue) {
+    return compareTextValues(airbyteMessageValue, destinationValue);
+  }
+
+  protected boolean compareTimeWithTimeZone(final String airbyteMessageValue, final String destinationValue) {
     return compareTextValues(airbyteMessageValue, destinationValue);
   }
 

airbyte-integrations/connectors/destination-gcs/src/test-integration/java/io/airbyte/integrations/destination/gcs/GcsAvroTestDataComparator.java

@@ -37,7 +37,7 @@ protected boolean compareDateTimeValues(String airbyteMessageValue, String desti
   }
 
   @Override
-  protected boolean compareTime(final String airbyteMessageValue, final String destinationValue) {
+  protected boolean compareTimeWithoutTimeZone(final String airbyteMessageValue, final String destinationValue) {
     LocalTime destinationDate = LocalTime.ofInstant(getInstantFromEpoch(destinationValue), ZoneOffset.UTC);
     LocalTime expectedDate = LocalTime.parse(airbyteMessageValue, DateTimeFormatter.ISO_TIME);
     return expectedDate.equals(destinationDate);

airbyte-integrations/connectors/destination-s3/src/test-integration/java/io/airbyte/integrations/destination/s3/S3AvroParquetTestDataComparator.java

@@ -41,7 +41,7 @@ protected boolean compareDateTimeValues(String airbyteMessageValue, String desti
   }
 
   @Override
-  protected boolean compareTime(final String airbyteMessageValue, final String destinationValue) {
+  protected boolean compareTimeWithoutTimeZone(final String airbyteMessageValue, final String destinationValue) {
     var destinationDate = LocalTime.ofInstant(getInstantFromEpoch(destinationValue), ZoneOffset.UTC);
     var expectedDate = LocalTime.parse(airbyteMessageValue, DateTimeFormatter.ISO_TIME);
     return expectedDate.equals(destinationDate);