Repository: microsoft/kafka-connect-cosmosdb
Pull request #385 (2021-05-16 21:45:58)

v1.0.6-beta
## Type of PR
pushing v1.0.6-beta


Patch:
diff --git a/.github/workflows/mavenCI.yml b/.github/workflows/mavenCI.yml
index 2c1e6a8..31fb90c 100644
--- a/.github/workflows/mavenCI.yml
+++ b/.github/workflows/mavenCI.yml
@@ -28,4 +28,9 @@ jobs:
     - name: Run Integration Test
       shell: pwsh
       run: ./src/docker/startup.ps1
-      if: github.ref == 'refs/heads/main'
\ No newline at end of file
+      if: github.ref == 'refs/heads/main'
+    - name: Upload JaCoCo coverage report
+      uses: actions/upload-artifact@v2
+      with:
+        name: jacoco-report
+        path: target/site/jacoco/
diff --git a/doc/README_Sink.md b/doc/README_Sink.md
index 9b63b18..c971dbc 100644
--- a/doc/README_Sink.md
+++ b/doc/README_Sink.md
@@ -176,7 +176,6 @@ The following settings are used to configure the Cosmos DB Kafka Sink Connector.
 | connect.cosmos.master.key | string | The Cosmos primary key that the sink connects with | Required |
 | connect.cosmos.databasename | string | The name of the Cosmos database the sink writes to | Required |
 | connect.cosmos.containers.topicmap | string | Mapping between Kafka Topics and Cosmos Containers, formatted using CSV as shown: `topic#container,topic2#container2` | Required |
-| connect.cosmos.sink.useUpsert | boolean | Whether to upsert items into Cosmos DB. Default is `false`. | Optional |
 | key.converter | string | Serialization format for the key data written into Kafka topic | Required |
 | value.converter | string | Serialization format for the value data written into the Kafka topic | Required |
 | key.converter.schemas.enable | string | Set to `"true"` if the key data has embedded schema | Optional |
diff --git a/pom.xml b/pom.xml
index 1f7e639..3f06950 100644
--- a/pom.xml
+++ b/pom.xml
@@ -7,7 +7,7 @@
 
     <groupId>com.azure.cosmos.kafka</groupId>
     <artifactId>kafka-connect-cosmos</artifactId>
-    <version>1.0.4-beta</version>
+    <version>1.0.6-beta</version>
 
     <name> kafka-connect-cosmos</name>
     <url>https://github.com/microsoft/kafka-connect-cosmosdb</url>
@@ -19,13 +19,14 @@
         <kafka.version>2.7.0</kafka.version>
         <log4j.version>2.14.0</log4j.version>
         <slf4j.version>1.7.30</slf4j.version>
+        <confluent-log4j.version>1.2.17-cp2</confluent-log4j.version>
     </properties>
 
     <repositories>
         <repository>
             <id>confluent</id>
             <name>Confluent</name>
-            <url>http://packages.confluent.io/maven/</url>
+            <url>https://packages.confluent.io/maven/</url>
         </repository>
     </repositories>
          
@@ -42,21 +43,12 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>connect-api</artifactId>
             <version>${kafka.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-clients</artifactId>
-            <version>${kafka.version}</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-cosmos</artifactId>
-            <version>4.11.0</version>
-        </dependency>
-        <dependency>
-            <groupId>io.confluent</groupId>
-            <artifactId>kafka-schema-registry-maven-plugin</artifactId>
-            <version>6.0.1</version>
+            <version>4.15.0</version>
         </dependency>
 
         <!-- Apache commons -->
@@ -87,6 +79,17 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>
             <version>${slf4j.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>log4j</groupId>
+                    <artifactId>log4j</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>io.confluent</groupId>
+            <artifactId>confluent-log4j</artifactId>
+            <version>${confluent-log4j.version}</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
@@ -315,6 +318,24 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <groupId>org.jacoco</groupId>
+                <artifactId>jacoco-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>prepare-agent</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>generate-code-coverage-report</id>
+                        <phase>test</phase>
+                        <goals>
+                            <goal>report</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
index ab3d86b..a083e11 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
@@ -8,8 +8,6 @@ import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.ConfigDef.Validator;
 import org.apache.kafka.common.config.ConfigDef.Width;
 
-import io.confluent.kafka.schemaregistry.utils.EnumRecommender;
-
 import java.util.Map;
 
 @SuppressWarnings ({"squid:S1854", "squid:S2160"})  // suppress unneeded int *groupOrder variables, equals method
@@ -37,13 +35,6 @@ public class CosmosDBConfig extends AbstractConfig {
     public static final int COSMOS_DATABASE_GROUP_ORDER = 2;
     public static final String COSMOS_CLIENT_USER_AGENT_SUFFIX = "APN/1.0 Microsoft/1.0 KafkaConnect/";
 
-    public static final EnumRecommender BOOLEAN_RECOMMENDER = EnumRecommender.in(BooleanValues.values());
-
-    public enum BooleanValues {
-        TRUE,
-        FALSE
-    }
-
     private String connEndpoint;
     private String connKey;
     private String databaseName;
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfig.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfig.java
index 7ac2fb9..d85282a 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfig.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfig.java
@@ -15,54 +15,11 @@ import com.azure.cosmos.kafka.connect.CosmosDBConfig;
 
 @SuppressWarnings ({"squid:S1854", "squid:S2160"})  // suppress unneeded int *groupOrder variables, equals method
 public class CosmosDBSinkConfig extends CosmosDBConfig {
-
-    static final String COSMOS_USE_UPSERT_CONF = "connect.cosmos.sink.useUpsert";
-    private static final String COSMOS_USE_UPSERT_DEFAULT = "false";
-    private static final String COSMOS_USE_UPSERT_DOC = "Behaviour of operation to Cosmos."
-        + " 'false' is the default value and signals that all operations to Cosmos are Insert;"
-        + " 'true' changes the behaviour to use Upsert operation.";
-    private static final String COSMOS_USE_UPSERT_DISPLAY = "Use Cosmos Upsert";
-
-    private String useUpsert;
-
     public CosmosDBSinkConfig(ConfigDef config, Map<String, String> parsedConfig) {
         super(config, parsedConfig);
     }
 
     public CosmosDBSinkConfig(Map<String, String> parsedConfig) {
         super(getConfig(), parsedConfig);
-
-        useUpsert = this.getString(COSMOS_USE_UPSERT_CONF);
-    }
-
-    public static ConfigDef getConfig() {
-        ConfigDef result = CosmosDBConfig.getConfig();
-        
-        defineDatabaseConfigs(result);
-
-        return result;
-    }
-
-    private static void defineDatabaseConfigs(ConfigDef result) {
-        final String databaseGroupName = "Database";
-        int databaseGroupOrder = CosmosDBConfig.COSMOS_DATABASE_GROUP_ORDER;
-        
-        result.define(
-            COSMOS_USE_UPSERT_CONF,
-            Type.STRING,
-            COSMOS_USE_UPSERT_DEFAULT,
-            BOOLEAN_RECOMMENDER,
-            Importance.MEDIUM,
-            COSMOS_USE_UPSERT_DOC,
-            databaseGroupName,
-            ++databaseGroupOrder,
-            Width.MEDIUM,
-            COSMOS_USE_UPSERT_DISPLAY,
-            BOOLEAN_RECOMMENDER
-        );
-    }
-  
-    public String getUseUpsert() {
-        return this.useUpsert;
     }
 }
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
index 28e0995..f06c656 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
@@ -90,11 +90,7 @@ public class CosmosDBSinkTask extends SinkTask {
     }
 
     private void addItemToContainer(CosmosContainer container, Object recordValue) {
-        if (config.getUseUpsert().equalsIgnoreCase(CosmosDBConfig.BooleanValues.TRUE.toString())) {
-            container.upsertItem(recordValue);
-        } else {
-            container.createItem(recordValue);
-        }
+        container.upsertItem(recordValue);
     }
 
     @Override
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfig.java b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfig.java
index ae9f9a8..09aa5bb 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfig.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfig.java
@@ -70,9 +70,9 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
     private Long bufferSize;
     private Long batchSize;
     private Long pollInterval;
-    private String messageKeyEnabled;
+    private boolean messageKeyEnabled;
     private String messageKeyField;
-    private String useLatestOffset;
+    private boolean useLatestOffset;
 
     // Variables not defined as Connect configs, should not be exposed when creating connector
     private String workerName;
@@ -89,9 +89,9 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
         bufferSize = this.getLong(COSMOS_SOURCE_TASK_BUFFER_SIZE_CONF);
         batchSize = this.getLong(COSMOS_SOURCE_TASK_BATCH_SIZE_CONF);
         pollInterval = this.getLong(COSMOS_SOURCE_TASK_POLL_INTERVAL_CONF);
-        messageKeyEnabled = this.getString(COSMOS_MESSAGE_KEY_ENABLED_CONF);
+        messageKeyEnabled = this.getBoolean(COSMOS_MESSAGE_KEY_ENABLED_CONF);
         messageKeyField = this.getString(COSMOS_MESSAGE_KEY_FIELD_CONF);
-        useLatestOffset = this.getString(COSMOS_USE_LATEST_OFFSET_CONF);
+        useLatestOffset = this.getBoolean(COSMOS_USE_LATEST_OFFSET_CONF);
 
         // Since variables are not defined as Connect configs, grab values directly from Map
         assignedContainer = parsedConfig.get(COSMOS_ASSIGNED_CONTAINER_CONF);
@@ -166,16 +166,14 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
         result
             .define(
                 COSMOS_USE_LATEST_OFFSET_CONF,
-                Type.STRING,
+                Type.BOOLEAN,
                 COSMOS_USE_LATEST_OFFSET_DEFAULT,
-                BOOLEAN_RECOMMENDER,
                 Importance.HIGH,
                 COSMOS_USE_LATEST_OFFSET_DOC,
                 databaseGroupName,
                 ++databaseGroupOrder,
                 Width.MEDIUM,
-                COSMOS_USE_LATEST_OFFSET_DISPLAY,
-                BOOLEAN_RECOMMENDER
+                COSMOS_USE_LATEST_OFFSET_DISPLAY
             );
     }
 
@@ -186,16 +184,14 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
         result
             .define(
                 COSMOS_MESSAGE_KEY_ENABLED_CONF,
-                Type.STRING,
+                Type.BOOLEAN,
                 COSMOS_MESSAGE_KEY_ENABLED_DEFAULT,
-                BOOLEAN_RECOMMENDER,
                 Importance.MEDIUM,
                 COSMOS_MESSAGE_KEY_ENABLED_DOC,
                 messageGroupName,
                 messageGroupOrder++,
                 Width.SHORT,
-                COSMOS_MESSAGE_KEY_ENABLED_DISPLAY,
-                BOOLEAN_RECOMMENDER
+                COSMOS_MESSAGE_KEY_ENABLED_DISPLAY
             )
             .define(
                 COSMOS_MESSAGE_KEY_FIELD_CONF,
@@ -227,7 +223,7 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
         return this.pollInterval;
     }
 
-    public String isMessageKeyEnabled() {
+    public boolean isMessageKeyEnabled() {
         return this.messageKeyEnabled;
     }
 
@@ -235,7 +231,7 @@ public class CosmosDBSourceConfig extends CosmosDBConfig {
         return this.messageKeyField;
     }
 
-    public String useLatestOffset() {
+    public boolean useLatestOffset() {
         return this.useLatestOffset;
     }
 
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceTask.java b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceTask.java
index c8350d7..666ff6c 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceTask.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceTask.java
@@ -70,7 +70,7 @@ public class CosmosDBSourceTask extends SourceTask {
         
         Map<String, Object> offset = context.offsetStorageReader().offset(partitionMap);
         // If NOT using the latest offset, reset lease container token to earliest possible value
-        if (config.useLatestOffset().equalsIgnoreCase(CosmosDBConfig.BooleanValues.FALSE.toString())) {
+        if (!config.useLatestOffset()) {
             updateContinuationToken(ZERO_CONTINUATION_TOKEN);
         } else if (offset != null) {
             // Check for previous offset and compare with lease container token
@@ -173,7 +173,7 @@ public class CosmosDBSourceTask extends SourceTask {
             try {                
                 // Set the Kafka message key if option is enabled and field is configured in document
                 String messageKey = "";
-                if (config.isMessageKeyEnabled().equalsIgnoreCase(CosmosDBConfig.BooleanValues.TRUE.toString())) {
+                if (config.isMessageKeyEnabled()) {
                     JsonNode messageKeyFieldNode = node.get(config.getMessageKeyField());
                     messageKey = (messageKeyFieldNode != null) ? messageKeyFieldNode.toString() : "";
                 }


Test patch:
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfigTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfigTest.java
index 3968382..b7f0d06 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfigTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConfigTest.java
@@ -31,13 +31,6 @@ public class CosmosDBSinkConfigTest {
         assertEquals("mycontainer", config.getTopicContainerMap().getContainerForTopic("mytopic").get());
     }
 
-    @Test
-    public void shouldHaveDefaultValues() {
-        // Adding required Configuration with no default value.
-        CosmosDBSinkConfig config = new CosmosDBSinkConfig(setupConfigs());
-        assertTrue(config.getUseUpsert().equalsIgnoreCase("false"));
-    }
-
     @Test
     public void shouldThrowExceptionWhenCosmosEndpointNotGiven() {
         // Adding required Configuration with no default value.
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorConfigTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorConfigTest.java
index a35efa9..463d8d2 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorConfigTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorConfigTest.java
@@ -32,15 +32,6 @@ public class CosmosDBSinkConnectorConfigTest {
             .get(CosmosDBSinkConfig.COSMOS_DATABASE_NAME_CONF));
     }
 
-    @Test
-    public void testPresentDefaults(){
-        //The task useUpsert has a default setting. Let's see if the configdef does
-        String useUpsert = new CosmosDBSinkConfig(CosmosDBSinkConfigTest.setupConfigs()).getUseUpsert();
-        assertNotNull(useUpsert);
-        assertEquals(useUpsert, new CosmosDBSinkConnector().config().defaultValues()
-            .get(CosmosDBSinkConfig.COSMOS_USE_UPSERT_CONF));
-    }
-
     @Test
     public void testTaskConfigs(){
         Map<String, String> settingAssignment = CosmosDBSinkConfigTest.setupConfigs();
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
index 5001739..6482ee1 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
@@ -42,7 +42,6 @@ public class CosmosDBSinkTaskTest {
         Map<String, String> settingAssignment = CosmosDBSinkConfigTest.setupConfigs();
         settingAssignment.put(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, topicName + "#" + containerName);
         settingAssignment.put(CosmosDBSinkConfig.COSMOS_DATABASE_NAME_CONF, databaseName);
-        settingAssignment.put(CosmosDBSinkConfig.COSMOS_USE_UPSERT_CONF, upsertFalse);
         CosmosDBSinkConfig config = new CosmosDBSinkConfig(settingAssignment);
         FieldUtils.writeField(testTask, "config", config, true);
 
@@ -64,7 +63,7 @@ public class CosmosDBSinkTaskTest {
         assertNotNull(record.value());
 
         //Make mock connector to serialize a non-JSON payload
-        when(mockContainer.createItem(any())).then((invocation) -> {
+        when(mockContainer.upsertItem(any())).then((invocation) -> {
             Object item = invocation.getArgument(0);
             //Will throw exception:
             try {
@@ -85,7 +84,7 @@ public class CosmosDBSinkTaskTest {
             fail("Expected ConnectException, but got: " + t.getClass().getName());
         }
 
-        verify(mockContainer, times(1)).createItem("foo");
+        verify(mockContainer, times(1)).upsertItem("foo");
     }
 
     @Test
@@ -98,7 +97,7 @@ public class CosmosDBSinkTaskTest {
         SinkRecord record = new SinkRecord(topicName, 1, stringSchema, "nokey", mapSchema, map, 0L);
         assertNotNull(record.value());
         testTask.put(Arrays.asList(record));
-        verify(mockContainer, times(1)).createItem(map);
+        verify(mockContainer, times(1)).upsertItem(map);
     }
 }
 
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
new file mode 100644
index 0000000..a4e102e
--- /dev/null
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
@@ -0,0 +1,96 @@
+package com.azure.cosmos.kafka.connect.sink;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+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.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Time;
+import org.apache.kafka.connect.data.Timestamp;
+import org.junit.Test;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class StructToJsonMapTest {
+    @Test
+    public void emptyStructToEmptyMap() {
+        Schema schema = SchemaBuilder.struct()
+            .build();
+        Struct struct = new Struct(schema);
+        assertEquals(ImmutableMap.of(), StructToJsonMap.toJsonMap(struct));
+    }
+
+    @Test
+    public void complextStructToMap() {
+        Schema embeddedSchema = SchemaBuilder.struct()
+            .field("embedded_string", Schema.STRING_SCHEMA)
+            .build();
+        Schema schema = SchemaBuilder.struct()
+            .field("boolean", Schema.BOOLEAN_SCHEMA)
+            .field("int8", Schema.INT8_SCHEMA)
+            .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("date", Date.SCHEMA)
+            .field("time", Time.SCHEMA)
+            .field("timestamp", Timestamp.SCHEMA)
+            .field("decimal", Decimal.schema(10))
+            .field("string", Schema.STRING_SCHEMA)
+            .field("struct", embeddedSchema)
+            .field("array_of_boolean", SchemaBuilder.array(Schema.BOOLEAN_SCHEMA).build())
+            .field("array_of_struct", SchemaBuilder.array(embeddedSchema).build())
+            .field("optional_string", Schema.OPTIONAL_STRING_SCHEMA)
+            .build();
+
+        java.util.Date now = new java.util.Date();
+        BigDecimal ten = new BigDecimal(10).setScale(10);
+        String quickBrownFox = "The quick brown fox jumps over the lazy dog";
+        Struct struct = new Struct(schema)
+            .put("boolean", false)
+            .put("int8", (byte) 0)
+            .put("int16", (short) 1)
+            .put("int32", 2)
+            .put("int64", (long) 3)
+            .put("float32", (float) 4.0)
+            .put("float64", (double) 5.0)
+            .put("date", now)
+            .put("time", now)
+            .put("timestamp", now)
+            .put("decimal", ten)
+            .put("string", quickBrownFox)
+            .put("struct", new Struct(embeddedSchema)
+                .put("embedded_string", quickBrownFox))
+            .put("array_of_boolean", ImmutableList.of(false))
+            .put("array_of_struct", ImmutableList.of(
+                new Struct(embeddedSchema).put("embedded_string", quickBrownFox)));
+
+        Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
+        assertEquals(false, converted.get("boolean"));
+        assertEquals((byte) 0, converted.get("int8"));
+        assertEquals((short) 1, converted.get("int16"));
+        assertEquals(2, converted.get("int32"));
+        assertEquals((long) 3, converted.get("int64"));
+        assertEquals(4.0f, (float) converted.get("float32"), 0.01f);
+        assertEquals(5.0d, (double) converted.get("float64"), 0.01d);
+        assertEquals(now, converted.get("date"));
+        assertEquals(now, converted.get("time"));
+        assertEquals(now, converted.get("timestamp"));
+        assertEquals(ten, converted.get("decimal"));
+        assertEquals(quickBrownFox, converted.get("string"));
+        assertEquals(quickBrownFox,
+            ((Map<String, Object>) converted.get("struct")).get("embedded_string"));
+        assertEquals(false, ((List<Boolean>) converted.get("array_of_boolean")).get(0));
+        assertEquals(ImmutableMap.of("embedded_string", quickBrownFox),
+            ((List<Struct>) converted.get("array_of_struct")).get(0));
+        assertNull(converted.get("optional_string"));
+    }
+}
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfigTest.java b/src/test/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfigTest.java
index 2f9d76b..a2e2dcf 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfigTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConfigTest.java
@@ -38,7 +38,7 @@ public class CosmosDBSourceConfigTest {
         assertEquals(10000L, config.getTaskBufferSize().longValue());
         assertEquals(100L, config.getTaskBatchSize().longValue());
         assertEquals(1000L, config.getTaskPollInterval().longValue());
-        assertTrue(config.useLatestOffset().equalsIgnoreCase("false"));
+        assertFalse(config.useLatestOffset());
     }
 
     @Test