Add Azure Cosmos DB sink

pom.xml

@@ -35,6 +35,11 @@
         <java.version>1.7</java.version>
     </properties>
     <dependencies>
+        <dependency>
+    <groupId>com.azure</groupId>
+    <artifactId>azure-core-http-netty</artifactId>
+    <version>1.14.2</version>
+</dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-api</artifactId>
@@ -53,7 +58,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.10.8</version>
+            <version>2.13.5</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
@@ -133,7 +138,20 @@
             <artifactId>iot-device-client</artifactId>
             <version>1.3.30</version>
         </dependency>
-
+        <!-- Microsoft -->
+        <dependency>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-sdk-bom</artifactId>
+            <version>1.2.23</version>
+            <type>pom</type>
+            <scope>import</scope>
+        </dependency>
+        <dependency>
+            <artifactId>azure-cosmos</artifactId>
+            <groupId>com.azure</groupId>
+            <version>4.59.0</version>
+        </dependency>
+
         <!-- AWS -->
         <dependency>
             <groupId>com.amazonaws</groupId>
@@ -149,7 +167,7 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-core</artifactId>
-            <version>2.9.7</version>
+            <version>2.13.5</version>
         </dependency>
         <dependency>
             <groupId>org.apache.pulsar</groupId>

src/main/java/net/acesinc/data/json/generator/JsonDataGenerator.java

@@ -109,6 +109,12 @@ public JsonDataGenerator(String simConfigString) {
                         }
                         break;
                     }
+                    // add a case for ComsoDB
+                    case "cosmosdb": {
+                        log.info("Adding CosmosDB Logger with properties: " + elProps);
+                        loggers.add(new CosmosDBLogger(elProps));
+                        break;
+                    }
                 }
             }
             if (loggers.isEmpty()) {

src/main/java/net/acesinc/data/json/generator/log/CosmosDBLogger.java

@@ -0,0 +1,56 @@
+package net.acesinc.data.json.generator.log;
+
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.CosmosItemResponse;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Map;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class CosmosDBLogger implements EventLogger {
+
+    private static final Logger log = LogManager.getLogger(CosmosDBLogger.class);
+    private CosmosClient cosmosClient;
+    private CosmosDatabase cosmosDatabase;
+    private CosmosContainer cosmosContainer;
+    private final ObjectMapper mapper = new ObjectMapper();
+
+    public CosmosDBLogger(Map<String, Object> props) {
+        String uri = (String) props.get("uri");
+        String key = (String) props.get("key");
+        String databaseName = (String) props.get("databaseName");
+        String containerName = (String) props.get("containerName");
+
+        cosmosClient = new CosmosClientBuilder()
+                .endpoint(uri)
+                .key(key)
+                .buildClient();
+
+        cosmosDatabase = cosmosClient.getDatabase(databaseName);
+        cosmosContainer = cosmosDatabase.getContainer(containerName);
+    }
+
+    @Override
+    public void logEvent(String event, Map<String, Object> producerConfig) {
+        try {
+            JsonNode jsonNode = mapper.readTree(event);
+            CosmosItemResponse<JsonNode> item = cosmosContainer.createItem(jsonNode);
+            log.info("Document added to Cosmos DB with request charge of " + item.getRequestCharge() + " within session " + item.getSessionToken());
+        } catch (Exception e) {
+            log.error("Error inserting JSON data into Cosmos DB", e);
+        }
+    }
+
+    @Override
+    public void shutdown() {
+        if (cosmosClient != null) {
+            cosmosClient.close();
+            log.info("Cosmos DB client closed successfully");
+        }
+    }
+}