Repository: microsoft/kafka-connect-cosmosdb (java)

Issue:
Nested Maps in Struct are not correctly parsed and lead to later error in writing the sink record
## Description
We currently have a heavily nested avro object that should be written with the SinkConnector to Cosmos DB. This process fails since the in the background used JacksonParser cannot handle the Struct object. The stracktrace looks like:

```
"trace" : "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:334)
\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:235)
\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:204)
\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:201)
\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:256)
\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
\tat java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.IllegalArgumentException: Failed to serialize the object into json
\tat com.azure.cosmos.implementation.Utils.serializeJsonToByteBuffer(Utils.java:702)
\tat com.azure.cosmos.implementation.InternalObjectNode.serializeJsonToByteBuffer(InternalObjectNode.java:116)
\tat com.azure.cosmos.BridgeInternal.serializeJsonToByteBuffer(BridgeInternal.java:105)
\tat com.azure.cosmos.implementation.RxDocumentClientImpl.getCreateDocumentRequest(RxDocumentClientImpl.java:1508)
\tat com.azure.cosmos.implementation.RxDocumentClientImpl.upsertDocumentInternal(RxDocumentClientImpl.java:1885)
\tat com.azure.cosmos.implementation.RxDocumentClientImpl.lambda$upsertDocument$39(RxDocumentClientImpl.java:1877)
\tat com.azure.cosmos.implementation.ObservableHelper.lambda$inlineIfPossibleAsObs$1(ObservableHelper.java:44)
\tat com.azure.cosmos.implementation.BackoffRetryUtility.lambda$executeRetry$0(BackoffRetryUtility.java:36)
\tat reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44)
\tat reactor.core.publisher.FluxRetryWhen.subscribe(FluxRetryWhen.java:77)
\tat reactor.core.publisher.MonoRetryWhen.subscribeOrReturn(MonoRetryWhen.java:46)
\tat reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:57)
\tat reactor.core.publisher.MonoDeferContextual.subscribe(MonoDeferContextual.java:55)
\tat reactor.core.publisher.Mono.subscribe(Mono.java:4490)
\tat reactor.core.publisher.Mono.block(Mono.java:1741)
\tat com.azure.cosmos.CosmosContainer.blockItemResponse(CosmosContainer.java:241)
\tat com.azure.cosmos.CosmosContainer.upsertItem(CosmosContainer.java:205)
\tat com.azure.cosmos.kafka.connect.sink.PointWriter.writeCore(PointWriter.java:31)
\tat com.azure.cosmos.kafka.connect.sink.SinkWriterBase.write(SinkWriterBase.java:26)
\tat com.azure.cosmos.kafka.connect.sink.CosmosDBSinkTask.put(CosmosDBSinkTask.java:137)
\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:584)
\t... 10 more
\tSuppressed: java.lang.Exception: #block terminated with an error
\t\tat reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:99)
\t\tat reactor.core.publisher.Mono.block(Mono.java:1742)
\t\t... 16 more
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: No serializer found for class org.apache.kafka.connect.data.Struct and no properties discovered to create BeanSerializer (to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS) (through reference chain: java.util.HashMap[\"FIELD\"]->java.util.HashMap[\"FIELD_BELOW\"]->java.util.HashMap[\"5629\"]->java.util.HashMap[\"3450\"])
\tat com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:77)
\tat com.fasterxml.jackson.databind.SerializerProvider.reportBadDefinition(SerializerProvider.java:1306)
\tat com.fasterxml.jackson.databind.DatabindContext.reportBadDefinition(DatabindContext.java:408)
\tat com.fasterxml.jackson.databind.ser.impl.UnknownSerializer.failForEmpty(UnknownSerializer.java:53)
\tat com.fasterxml.jackson.databind.ser.impl.UnknownSerializer.serialize(UnknownSerializer.java:30)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeFields(MapSerializer.java:808)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeWithoutTypeInfo(MapSerializer.java:764)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:720)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:35)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeFields(MapSerializer.java:808)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeWithoutTypeInfo(MapSerializer.java:764)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:720)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:35)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeFields(MapSerializer.java:808)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeWithoutTypeInfo(MapSerializer.java:764)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:720)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:35)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeFields(MapSerializer.java:808)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeWithoutTypeInfo(MapSerializer.java:764)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:720)
\tat com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:35)
\tat com.fasterxml.jackson.databind.ser.DefaultSerializerProvider._serialize(DefaultSerializerProvider.java:480)
\tat com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:319)
\tat com.fasterxml.jackson.databind.ObjectMapper._writeValueAndClose(ObjectMapper.java:4624)
\tat com.fasterxml.jackson.databind.ObjectMapper.writeValue(ObjectMapper.java:3828)
\tat com.azure.cosmos.implementation.Utils.serializeJsonToByteBuffer(Utils.java:698)
```
## Possible working Solution (tested)
I could not make a easy way to reproduce this error but i''ll try to provide it as soon as possible. A solution i listed already in another bug report worked though. I changed the StructToJsonMap

```
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package com.azure.cosmos.kafka.connect.sink;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.json.JsonConverter;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

public class StructToJsonMap {

    public static Map<String, Object> toJsonMap(Struct struct) {
        Map<String, Object> jsonMap = new HashMap<>();

        JsonConverter jsonConverter = new JsonConverter();
        jsonConverter.configure(
                new HashMap<>(){{
                    put("schemas.enable", false);
                    put("decimal.format", "NUMERIC");
                    put("json.date.format", "yyyy-MM-dd'T'HH:mm:ss.SSSXXX");
                }}, false);

        byte[] payload = jsonConverter.fromConnectData("", struct.schema(), struct);
        String json = new String(payload, StandardCharsets.UTF_8);
        ObjectMapper mapper = new ObjectMapper();
        try {
            jsonMap = mapper.readValue(json, Map.class);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }

        return jsonMap;
    }

}
````

Solution looks bit hacky at first but i think its fine and it gives some huge benefits:
- Properties for Converter could be provided from the Connector configuration
- Data written to Cosmos DB could be configured easily for the user (if certain dates are prefered or decimal format)
- Converter offers even some more options
- Complexity in the code is replaced by already well tested code piece


This solution needs better error handling in case the ObjectMapper fails or the return value could just be plain json string (so just remove the Object Mapper). This code did not went through the style check. For testing i removed the plugins in the pom to build the project.


Data itself looks good in Cosmos DB:

```
    "someField": {
        "someField1": {
            "someField2": "eguy0yuv7i",
            "someField3": 4709,
            "someField4": "cx1gpmmnlo",
            "someField5": "gdnrvzmwp5",
            "someField6": "eiw4azchy6",
            "someField7": 1680011226474,
            "someField8": -2598,
            "someField9": {
                "levelLower1": 0.32,
                "levelLower2": 0.2,
                "levelLower3": 0.89,
                "levelLower4": 0.29,
                "levelLower5": 0.69,
                "levelLower6": 0.96
```

I'd really suggest this change since it's a really easy solution and might even fix unknown bugs.












Pull request description:
Fixed issue with maps with nested struct types for sink task
* Fixes #507 
* Fixes security vulnerability in json-path version
* Updated azure-cosmos to 4.42.0

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 37ab131..44d2fc2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,14 @@
 ## Release History
 
+### 1.8.0 (Unreleased)
+#### New Features
+* Updated `azure-cosmos` version to 4.42.0.
+* Updated `json-path` version to 2.8.0.
+* Updated `kafka-connect` version to 2.8.2.
+
+#### Key Bug Fixes
+* Fixed issue in Sink Connector for nested maps in Struct types. [PR 510](https://github.com/microsoft/kafka-connect-cosmosdb/pull/510)
+
 ### 1.7.1 (2023-03-24)
 #### Key Bug Fixes
 * Fixed issue in Sink Connector when using `AvroConverter`. [PR 503](https://github.com/microsoft/kafka-connect-cosmosdb/pull/503)
diff --git a/pom.xml b/pom.xml
index 9fa6582..f93b85d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,7 +16,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>11</maven.compiler.source>
         <maven.compiler.target>11</maven.compiler.target>
-        <kafka.version>2.8.1</kafka.version>
+        <kafka.version>2.8.2</kafka.version>
         <log4j.version>2.17.1</log4j.version>
         <slf4j.version>1.7.30</slf4j.version>
         <confluent-log4j.version>1.2.17-cp2</confluent-log4j.version>
@@ -48,12 +48,12 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-cosmos</artifactId>
-            <version>4.41.0</version>
+            <version>4.42.0</version>
         </dependency>
         <dependency>
             <groupId>com.jayway.jsonpath</groupId>
             <artifactId>json-path</artifactId>
-            <version>2.6.0</version>
+            <version>2.8.0</version>
         </dependency>
 
         <!-- Apache commons -->
@@ -65,7 +65,7 @@
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
-            <version>3.11</version>
+            <version>3.12.0</version>
         </dependency>
 
         <!--Logging-->
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 c5e24ec..acd7d01 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
@@ -112,6 +112,8 @@ public class CosmosDBSinkTask extends SinkTask {
                 if (record.value() instanceof Struct) {
                     recordValue = StructToJsonMap.toJsonMap((Struct) record.value());
                     //  TODO: Do we need to update the value schema to map or keep it struct?
+                } else if (record.value() instanceof Map) {
+                    recordValue = StructToJsonMap.handleMap((Map<String, Object>) record.value());
                 } else {
                     recordValue = record.value();
                 }
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
index 2b515b7..3a61b45 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMap.java
@@ -73,6 +73,9 @@ public class StructToJsonMap {
                 case STRUCT:
                     jsonMap.put(fieldName, toJsonMap(struct.getStruct(fieldName)));
                     break;
+                case MAP:
+                    jsonMap.put(fieldName, handleMap(struct.getMap(fieldName)));
+                    break;
                 default:
                     jsonMap.put(fieldName, struct.get(fieldName));
                     break;
@@ -80,4 +83,34 @@ public class StructToJsonMap {
         }
         return jsonMap;
     }
+
+    public static Map<String, Object> handleMap(Map<String, Object> map) {
+        if (map == null) {
+            return null;
+        }
+        Map<String, Object> cacheMap = new HashMap<>();
+        map.forEach((key, value) -> {
+            if (value instanceof Map) {
+                cacheMap.put(key, handleMap((Map<String, Object>) value));
+            } else if (value instanceof Struct) {
+                cacheMap.put(key, toJsonMap((Struct) value));
+            } else if (value instanceof List) {
+                List<Object> list = (List<Object>) value;
+                List<Object> jsonArray = new ArrayList<>();
+                list.forEach(item -> {
+                    if (item instanceof Struct) {
+                        jsonArray.add(toJsonMap((Struct) item));
+                    } else if (item instanceof Map) {
+                        jsonArray.add(handleMap((Map<String, Object>) item));
+                    } else {
+                        jsonArray.add(item);
+                    }
+                });
+                cacheMap.put(key, jsonArray);
+            } else {
+                cacheMap.put(key, value);
+            }
+        });
+        return cacheMap;
+    }
 }


Test patch:
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 f531469..6e1ad1a 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
@@ -244,6 +244,69 @@ public class CosmosDBSinkTaskTest {
         }
     }
 
+    @Test
+    public void sinkWriteSucceededWithMapRecordValue() {
+        Schema stringSchema = new ConnectSchema(Schema.Type.STRING);
+        Schema mapSchema = new ConnectSchema(Schema.Type.MAP);
+        Map<String, String> map = new HashMap<>();
+        map.put("foo", "fooz");
+        map.put("bar", "baaz");
+
+
+        SinkRecord record = new SinkRecord(topicName, 1, stringSchema, "nokey", mapSchema, map, 0L);
+        assertNotNull(record.value());
+
+        SinkWriteResponse sinkWriteResponse = new SinkWriteResponse();
+        sinkWriteResponse.getSucceededRecords().add(record);
+
+        MockedConstruction<? extends SinkWriterBase> mockedWriterConstruction = null;
+        AtomicReference<List<SinkRecord>> sinkRecords = new AtomicReference<>();
+        try {
+            if (this.isBulkModeEnabled) {
+                mockedWriterConstruction = mockConstructionWithAnswer(BulkWriter.class, invocation -> {
+                    if (invocation.getMethod().equals(BulkWriter.class.getMethod("write", List.class))) {
+                        sinkRecords.set(invocation.getArgument(0));
+                        return sinkWriteResponse;
+                    }
+
+                    throw new IllegalStateException("Not implemented for method " + invocation.getMethod().getName());
+                });
+            } else {
+                mockedWriterConstruction = mockConstructionWithAnswer(PointWriter.class, invocation -> {
+                    if (invocation.getMethod().equals(PointWriter.class.getMethod("write", List.class))) {
+                        sinkRecords.set(invocation.getArgument(0));
+                        return sinkWriteResponse;
+                    }
+
+                    throw new IllegalStateException("Not implemented for method " + invocation.getMethod().getName());
+                });
+            }
+
+            try {
+                testTask.put(List.of(record));
+            } catch (ConnectException ce) {
+                fail("Expected sink write succeeded. but got: " + ce.getMessage());
+            } catch (Throwable t) {
+                fail("Expected sink write succeeded, but got: " + t.getClass().getName());
+            }
+
+            assertEquals(1, mockedWriterConstruction.constructed().size());
+
+            SinkRecord sinkRecord = sinkRecords.get().get(0);
+            assertRecordEquals(record, sinkRecord);
+
+            Object value = sinkRecord.value();
+            assertTrue(value instanceof Map);
+
+            assertEquals("fooz", ((Map<?, ?>) value).get("foo"));
+            assertEquals("baaz", ((Map<?, ?>) value).get("bar"));
+        } finally {
+            if (mockedWriterConstruction != null) {
+                mockedWriterConstruction.close();
+            }
+        }
+    }
+
     private void assertRecordEquals(SinkRecord record, SinkRecord updatedRecord) {
         assertEquals(record.kafkaOffset(), updatedRecord.kafkaOffset());
         assertEquals(record.timestamp(), updatedRecord.timestamp());
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
index b0e6350..859158f 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/StructToJsonMapTest.java
@@ -3,6 +3,8 @@
 
 package com.azure.cosmos.kafka.connect.sink;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.connect.data.Date;
 import org.apache.kafka.connect.data.Decimal;
 import org.apache.kafka.connect.data.Schema;
@@ -13,6 +15,7 @@ import org.apache.kafka.connect.data.Timestamp;
 import org.junit.Test;
 
 import java.math.BigDecimal;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -24,7 +27,7 @@ public class StructToJsonMapTest {
     @Test
     public void emptyStructToEmptyMap() {
         Schema schema = SchemaBuilder.struct()
-                .build();
+                                     .build();
         Struct struct = new Struct(schema);
         assertEquals(Map.of(), StructToJsonMap.toJsonMap(struct));
     }
@@ -33,10 +36,10 @@ public class StructToJsonMapTest {
     @Test
     public void structWithEmptyArrayToMap() {
         Schema schema = SchemaBuilder.struct()
-                .field("array_of_boolean", SchemaBuilder.array(Schema.BOOLEAN_SCHEMA).build());
+                                     .field("array_of_boolean", SchemaBuilder.array(Schema.BOOLEAN_SCHEMA).build());
 
         Struct struct = new Struct(schema)
-                .put("array_of_boolean", List.of());
+            .put("array_of_boolean", List.of());
 
         Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
         assertEquals(List.of(), converted.get("array_of_boolean"));
@@ -45,48 +48,51 @@ public class StructToJsonMapTest {
     @Test
     public void complexStructToMap() {
         Schema embeddedSchema = SchemaBuilder.struct()
-                .field("embedded_string", Schema.STRING_SCHEMA)
-                .build();
+                                             .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();
+                                     .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)
+                                     .field("map",
+                                         SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).build())
+                                     .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", List.of(false))
-                .put("array_of_struct", List.of(
-                        new Struct(embeddedSchema).put("embedded_string", quickBrownFox)));
+            .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", List.of(false))
+            .put("array_of_struct", List.of(
+                new Struct(embeddedSchema).put("embedded_string", quickBrownFox)))
+            .put("map", Map.of("key", "value"));
 
         Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
         assertEquals(false, converted.get("boolean"));
@@ -103,29 +109,199 @@ public class StructToJsonMapTest {
         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(Map.of("embedded_string", quickBrownFox), ((List<Struct>) converted.get("array_of_struct")).get(0));
+        assertEquals(Map.of("embedded_string", quickBrownFox),
+            ((List<Struct>) converted.get("array_of_struct")).get(0));
         assertNull(converted.get("optional_string"));
+        assertEquals(Map.of("key", "value"), converted.get("map"));
     }
 
     @Test
     public void nullableStructTest() {
         Schema embeddedSchema = SchemaBuilder.struct()
-                .optional()
-                .field("embedded_string", Schema.STRING_SCHEMA)
-                .field("embedded_int", Schema.INT32_SCHEMA)
-                .build();
+                                             .optional()
+                                             .field("embedded_string", Schema.STRING_SCHEMA)
+                                             .field("embedded_int", Schema.INT32_SCHEMA)
+                                             .build();
         Schema schema = SchemaBuilder.struct()
-                .field("struct", embeddedSchema)
-                .field("string", Schema.STRING_SCHEMA)
-                .build();
+                                     .field("struct", embeddedSchema)
+                                     .field("string", Schema.STRING_SCHEMA)
+                                     .build();
 
         Struct struct = new Struct(schema)
-                .put("struct", null)
-                .put("string", "string_value");
+            .put("struct", null)
+            .put("string", "string_value");
 
         Map<String, Object> converted = StructToJsonMap.toJsonMap(struct);
         assertEquals("string_value", converted.get("string"));
         assertNull(converted.get("struct"));
     }
 
+    @Test
+    public void structWithMap() throws JsonProcessingException {
+        Schema schemaWithMapNested = SchemaBuilder.struct()
+                                                  .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA,
+                                                                                 Schema.STRING_SCHEMA)
+                                                                             .build()
+                                                  ).build();
+
+        Struct structWithMapNested = new Struct(schemaWithMapNested)
+            .put("map", Map.of("key", "value"));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.toJsonMap(structWithMapNested);
+        assertEquals(Map.of("key", "value"), toJsonMap.get("map"));
+        assertEquals("{\"map\":{\"key\":\"value\"}}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", Map.of("key", "value")), toJsonMap);
+    }
+
+    @Test
+    public void structWithMapWithStruct() throws JsonProcessingException {
+        Schema schemaWithMapNested = SchemaBuilder.struct()
+                                                  .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA,
+                                                      SchemaBuilder.struct()
+                                                                   .field("key", Schema.STRING_SCHEMA)
+                                                                   .build()
+                                                  ).build())
+                                                  .build();
+
+        Struct structWithMapNested = new Struct(schemaWithMapNested)
+            .put("map", Map.of("key", new Struct(schemaWithMapNested.field("map").schema().valueSchema())
+                .put("key", "value")));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.toJsonMap(structWithMapNested);
+        assertEquals(Map.of("key", Map.of("key", "value")), toJsonMap.get("map"));
+        assertEquals("{\"map\":{\"key\":{\"key\":\"value\"}}}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", Map.of("key", Map.of("key", "value"))), toJsonMap);
+    }
+
+    @Test
+    public void structWithMapWithList() throws JsonProcessingException {
+        Schema schemaWithMapNested = SchemaBuilder.struct()
+                                                  .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA,
+                                                      SchemaBuilder.struct()
+                                                                   .field("key",
+                                                                       SchemaBuilder.array(Schema.STRING_SCHEMA).build())
+                                                                   .build()
+                                                  ).build())
+                                                  .build();
+
+        Struct structWithMapNested = new Struct(schemaWithMapNested)
+            .put("map", Map.of("key", new Struct(schemaWithMapNested.field("map").schema().valueSchema())
+                .put("key", List.of("value"))));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.toJsonMap(structWithMapNested);
+        assertEquals(Map.of("key", Map.of("key", List.of("value"))), toJsonMap.get("map"));
+        assertEquals("{\"map\":{\"key\":{\"key\":[\"value\"]}}}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", Map.of("key", Map.of("key", List.of("value")))), toJsonMap);
+    }
+
+    @Test
+    public void mapWithValue() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        map.put("key", "value");
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals("value", toJsonMap.get("key"));
+        assertEquals(Map.of("key", "value"), toJsonMap);
+        assertEquals("{\"key\":\"value\"}", new ObjectMapper().writeValueAsString(toJsonMap));
+
+        map = new HashMap<>();
+        map.put("key", 10);
+        toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(10, toJsonMap.get("key"));
+        assertEquals(Map.of("key", 10), toJsonMap);
+        assertEquals("{\"key\":10}", new ObjectMapper().writeValueAsString(toJsonMap));
+    }
+
+    @Test
+    public void mapWithMap() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        map.put("map", Map.of("key", Map.of("key", Map.of("key1", "value1"))));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(Map.of("key", Map.of("key", Map.of("key1", "value1"))), toJsonMap.get("map"));
+        assertEquals("{\"map\":{\"key\":{\"key\":{\"key1\":\"value1\"}}}}",
+            new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", Map.of("key", Map.of("key", Map.of("key1", "value1")))), toJsonMap);
+    }
+
+    @Test
+    public void mapWithStruct() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        Struct struct = new Struct(SchemaBuilder.struct().field("key", Schema.STRING_SCHEMA).build())
+            .put("key", "value");
+        map.put("map", struct);
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(Map.of("key", "value"), toJsonMap.get("map"));
+        assertEquals("{\"map\":{\"key\":\"value\"}}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", Map.of("key", "value")), toJsonMap);
+    }
+
+    @Test
+    public void mapWithStructWithMap() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        Schema schemaWithMapNested = SchemaBuilder.struct()
+                                                  .field("map", SchemaBuilder.map(Schema.STRING_SCHEMA,
+                                                                                 Schema.STRING_SCHEMA)
+                                                                             .build()
+                                                  ).build();
+
+        Struct structWithMapNested = new Struct(schemaWithMapNested)
+            .put("map", Map.of("key", "value"));
+        map.put("struct", structWithMapNested);
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(Map.of("map", Map.of("key", "value")), toJsonMap.get("struct"));
+        assertEquals("{\"struct\":{\"map\":{\"key\":\"value\"}}}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("struct", Map.of("map", Map.of("key", "value"))), toJsonMap);
+    }
+
+    @Test
+    public void mapWithList() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        map.put("map", List.of("value1", "value2"));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(List.of("value1", "value2"), toJsonMap.get("map"));
+        assertEquals("{\"map\":[\"value1\",\"value2\"]}", new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", List.of("value1", "value2")), toJsonMap);
+    }
+
+    @Test
+    public void mapWithListWithMap() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        map.put("map", List.of(Map.of("key", "value1"), Map.of("key", "value2")));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(List.of(Map.of("key", "value1"), Map.of("key", "value2")), toJsonMap.get("map"));
+        assertEquals("{\"map\":[{\"key\":\"value1\"},{\"key\":\"value2\"}]}",
+            new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", List.of(Map.of("key", "value1"), Map.of("key", "value2"))), toJsonMap);
+    }
+
+    @Test
+    public void mapWithListWithStruct() throws JsonProcessingException {
+        Map<String, Object> map = new HashMap<>();
+        Struct struct1 = new Struct(SchemaBuilder.struct().field("key1", Schema.STRING_SCHEMA).build())
+            .put("key1", "value1");
+        Struct struct2 = new Struct(SchemaBuilder.struct().field("key2", Schema.STRING_SCHEMA).build())
+            .put("key2", "value2");
+        map.put("map", List.of(struct1, struct2));
+
+
+        Map<String, Object> toJsonMap = StructToJsonMap.handleMap(map);
+        assertEquals(List.of(Map.of("key1", "value1"), Map.of("key2", "value2")), toJsonMap.get("map"));
+        assertEquals("{\"map\":[{\"key1\":\"value1\"},{\"key2\":\"value2\"}]}",
+            new ObjectMapper().writeValueAsString(toJsonMap));
+        assertEquals(Map.of("map", List.of(Map.of("key1", "value1"), Map.of("key2", "value2"))), toJsonMap);
+    }
+
 }