If value is null, delete document in solr

Author: bkatwal

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

@@ -49,7 +49,7 @@ public void start(Map<String, String> configMap) {
     }
 
     @Override
-    public void put(Collection<SinkRecord> kafkaRecords) {
+    public void put(Collection<SinkRecord> kafkaRecords){
 
         for (SinkRecord record : kafkaRecords) {
             String id = record.key() != null ? record.key().toString() : null;
@@ -61,20 +61,16 @@ public void put(Collection<SinkRecord> kafkaRecords) {
             //TODO handle schema based record later
             if (valueSchema == null) {
 
-                if (record.value() == null) {
-                    log.error("No value passed for doc ID, {}", id);
-                    continue;
-                }
-
                 Map<String, Object> jsonValueMap = (Map<String, Object>) record.value();
 
                 Object delVal = jsonValueMap.get("_delete_");
 
                 //delete the field "_delete_" after reading the value from it
                 jsonValueMap.remove("_delete_");
 
-                //if _delete_ is passed in doc and is false, will try to delete doc
-                if (isDeleteRequest(delVal)) {
+                //if _delete_ field is passed as false in value or if value is null, respective
+                // doc will be deleted from solr
+                if (isDeleteRequest(delVal) || record.value() == null) {
 
                     sinkService.deleteById(id);