Remove custom debezium mssql converted for datetimeoffset type (#39342)

Author: rodireich

airbyte-integrations/connectors/source-mssql/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
-  dockerImageTag: 4.0.27
+  dockerImageTag: 4.0.28
   dockerRepository: airbyte/source-mssql
   documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
   githubIssueLabel: source-mssql

airbyte-integrations/connectors/source-mssql/src/main/java/io/airbyte/integrations/source/mssql/MssqlDebeziumConverter.java

@@ -17,7 +17,6 @@
 import java.time.*;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
-import microsoft.sql.DateTimeOffset;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,7 +33,6 @@ public class MssqlDebeziumConverter implements CustomConverter<SchemaBuilder, Re
   private static final String SMALLMONEY_TYPE = "SMALLMONEY";
   private static final String GEOMETRY = "GEOMETRY";
   private static final String GEOGRAPHY = "GEOGRAPHY";
-  private static final String DEBEZIUM_DATETIMEOFFSET_FORMAT = "yyyy-MM-dd HH:mm:ss[.][SSSSSSS] XXX";
 
   private static final String DATETIME_FORMAT_MICROSECONDS = "yyyy-MM-dd'T'HH:mm:ss[.][SSSSSS]";
 
@@ -58,8 +56,6 @@ public void converterFor(final RelationalColumn field,
       registerGeography(field, registration);
     } else if (TIME_TYPE.equalsIgnoreCase(field.typeName())) {
       registerTime(field, registration);
-    } else if (DATETIMEOFFSET.equalsIgnoreCase(field.typeName())) {
-      registerDateTimeOffSet(field, registration);
     }
   }
 
@@ -144,25 +140,6 @@ private void registerDatetime(final RelationalColumn field,
 
   }
 
-  private void registerDateTimeOffSet(final RelationalColumn field,
-                                      final ConverterRegistration<SchemaBuilder> registration) {
-    registration.register(SchemaBuilder.string(), input -> {
-      if (Objects.isNull(input)) {
-        return DebeziumConverterUtils.convertDefaultValue(field);
-      }
-
-      if (input instanceof DateTimeOffset) {
-        return DataTypeUtils.toISO8601String(
-            OffsetDateTime.parse(input.toString(),
-                DateTimeFormatter.ofPattern(DEBEZIUM_DATETIMEOFFSET_FORMAT)));
-      }
-
-      LOGGER.warn("Uncovered DateTimeOffSet class type '{}'. Use default converter",
-          input.getClass().getName());
-      return input.toString();
-    });
-  }
-
   private void registerTime(final RelationalColumn field,
                             final ConverterRegistration<SchemaBuilder> registration) {
     registration.register(SchemaBuilder.string(), input -> {

airbyte-integrations/connectors/source-mssql/src/test-integration/java/io/airbyte/integrations/source/mssql/AbstractMssqlSourceDatatypeTest.java

@@ -183,9 +183,11 @@ protected void initTests() {
         TestDataHolder.builder()
             .sourceType("datetimeoffset")
             .airbyteType(JsonSchemaType.STRING_TIMESTAMP_WITH_TIMEZONE)
-            .addInsertValues("'0001-01-10 00:00:00 +01:00'", "'9999-01-10 00:00:00 +01:00'", "null")
+            .addInsertValues("'0001-01-10 00:00:00 +01:00'", "'9999-01-10 00:00:00 +01:00'", "null", "'2024-05-10 19:00:01.604805 +03:00'",
+                "'2024-03-02 19:08:07.1234567 +09:00'", "'2024-03-02 19:08:07.12345678 +09:00'")
             .addExpectedValues("0001-01-10 00:00:00.0000000 +01:00",
-                "9999-01-10 00:00:00.0000000 +01:00", null)
+                "9999-01-10 00:00:00.0000000 +01:00", null, "2024-05-10 19:00:01.6048050 +03:00", "2024-03-02 19:08:07.1234567 +09:00",
+                "2024-03-02 19:08:07.1234568 +09:00")
             .createTablePatternSql(CREATE_TABLE_SQL)
             .build());