dynamic fields support

Author: bkatwal

manifest.json

@@ -53,5 +53,5 @@
 
   "title": "Kafka Connect solr sink",
 
-  "version": "1.0"
+  "version": "2.0"
 }

src/main/java/com/bkatwal/kafkaproject/SolrSinkConnectorConfig.java

@@ -14,6 +14,8 @@
 
 package com.bkatwal.kafkaproject;
 
+import static com.bkatwal.kafkaproject.utils.SolrMode.CLOUD;
+
 import com.bkatwal.kafkaproject.utils.SolrMode;
 import java.util.Map;
 import org.apache.kafka.common.config.AbstractConfig;
@@ -59,7 +61,7 @@ public static ConfigDef conf() {
         .define(SOLRURL_CONFIG, Type.STRING, Importance.HIGH, SOLRURL_DOC)
         .define(USERNAME_CONFIG, Type.STRING, "", Importance.MEDIUM, USERNAME_DOC)
         .define(PASSWORD_CONFIG, Type.PASSWORD, "", Importance.MEDIUM, PASSWORD_DOC)
-        .define(SOLRMODE_CONFIG, Type.STRING, "", Importance.MEDIUM, SOLRMODE_DOC)
+        .define(SOLRMODE_CONFIG, Type.STRING, CLOUD.name(), Importance.MEDIUM, SOLRMODE_DOC)
         .define(COMMIT_WITHIN_MS, Type.INT, 10, Importance.MEDIUM, COMMIT_WITHIN_MS_DOC);
   }
 

src/main/java/com/bkatwal/kafkaproject/utils/PlainJsonSolrDocMappersImpl.java

@@ -17,8 +17,10 @@
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.solr.common.SolrInputDocument;
@@ -52,7 +54,7 @@ public SolrInputDocument convertToSolrDocument(SinkRecord sinkRecord) {
 
     Map<String, Object> obj = (Map<String, Object>) sinkRecord.value();
 
-    return toSolrDoc(obj);
+    return toSolrDoc(createDynamicFieldsForRecordIfExists(obj));
   }
 
   /*
@@ -75,6 +77,29 @@ private void addFieldsToDoc(Map<String, Object> objectMap, SolrInputDocument doc
 
   }
 
+  private Map<String, Object> createDynamicFieldsForRecordIfExists(
+      final Map<String, Object> record) {
+
+    Map<String, Object> newRecord = new LinkedHashMap<>();
+
+    for (Entry<String, Object> entry : record.entrySet()) {
+
+      String entryKey = entry.getKey();
+      Object entryValue = entry.getValue();
+
+      if (entryValue instanceof Map) {
+        Map<String, Object> columnVal = (Map<String, Object>) entry.getValue();
+        String columnName = entry.getKey();
+        columnVal.forEach(
+            (key, val) -> newRecord.put(columnName.concat("_").concat(key), val));
+      } else {
+        newRecord.put(entryKey, entryValue);
+      }
+    }
+    record.clear();
+    return newRecord;
+  }
+
   private Collection<SolrInputDocument> getChildDocuments(Object childDocuments) {
 
     List<Map<String, Object>> childDocsList = (List<Map<String, Object>>) childDocuments;

src/main/java/com/bkatwal/kafkaproject/utils/SolrSinkService.java

@@ -47,7 +47,7 @@ public boolean update(final String id, final SinkRecord data) {
   public boolean deleteById(final String id) {
     UpdateResponse updateResponse;
     try {
-      updateResponse = solrClient.deleteById(collection, id, 10);
+      updateResponse = solrClient.deleteById(collection, id, commitWithinMs);
     } catch (SolrServerException | IOException e) {
       log.error("Unable to send delete request to solr");
       throw new SolrException(ErrorCode.SERVER_ERROR, "Unable to send delete request to solr", e);
@@ -61,7 +61,7 @@ public boolean insert(final String id, final SinkRecord record) {
     UpdateResponse updateResponse;
     SolrInputDocument solrInputDocument = jsonSolrDocMapper.convertToSolrDocument(record);
     try {
-      updateResponse = solrClient.add(collection, solrInputDocument, 10);
+      updateResponse = solrClient.add(collection, solrInputDocument, commitWithinMs);
       log.debug("saved document: {}", solrInputDocument);
     } catch (SolrServerException | IOException e) {
       log.error("Unable to send update request to solr");