Ensure BsonString value is passed to the FlexibleDateTimeParser

Author: rozza

src/integrationTest/java/com/mongodb/kafka/connect/sink/MongoSinkTaskIntegrationTest.java

@@ -102,9 +102,9 @@ void testSinkProcessesTimeseriesData() {
                   i -> {
                     Date now = new Date();
                     Document doc = new Document("_id", i);
-                    if (i == 0) {
-                      doc.put("ts", "1970T01:01:01.000001Z");
-                    } else if (i == 1) {
+                    if (i == 1) {
+                      doc.put("ts", "1970-01-01T01:01:01.000001Z");
+                    } else if (i == 2) {
                       doc.put("ts", 3600000);
                     } else {
                       doc.put("ts", now);

src/main/java/com/mongodb/kafka/connect/sink/processor/TimeseriesTimeFieldAutoConversion.java

@@ -67,7 +67,7 @@ public void process(final SinkDocument doc, final SinkRecord orig) {
                         () ->
                             new BsonDateTime(
                                 flexibleDateTimeParser.toEpochMilli(
-                                    timeField.asString().toString())));
+                                    timeField.asString().getValue())));
               }
               convertedValue.map(bsonDateTime -> d.put(fieldName, bsonDateTime));
             });