Improve MongoDB CDC debezium error messages

Author: rozza

src/main/java/com/mongodb/kafka/connect/sink/cdc/debezium/mongodb/MongoDbDelete.java

@@ -41,6 +41,10 @@ public WriteModel<BsonDocument> perform(final SinkDocument doc) {
             .orElseThrow(
                 () -> new DataException("Key document must not be missing for delete operation"));
 
+    if (!keyDoc.containsKey(JSON_ID_FIELD)) {
+      throw new DataException(format("Delete document missing `%s` field.", JSON_ID_FIELD));
+    }
+
     try {
       return new DeleteOneModel<>(
           BsonDocument.parse(

src/main/java/com/mongodb/kafka/connect/sink/cdc/debezium/mongodb/MongoDbInsert.java

@@ -19,6 +19,7 @@
 package com.mongodb.kafka.connect.sink.cdc.debezium.mongodb;
 
 import static com.mongodb.kafka.connect.sink.cdc.debezium.mongodb.MongoDbHandler.ID_FIELD;
+import static java.lang.String.format;
 
 import org.apache.kafka.connect.errors.DataException;
 
@@ -44,6 +45,10 @@ public WriteModel<BsonDocument> perform(final SinkDocument doc) {
             .orElseThrow(
                 () -> new DataException("Value document must not be missing for insert operation"));
 
+    if (!valueDoc.containsKey(JSON_DOC_FIELD_PATH)) {
+      throw new DataException(format("Insert document missing `%s` field.", JSON_DOC_FIELD_PATH));
+    }
+
     try {
       BsonDocument insertDoc =
           BsonDocument.parse(valueDoc.get(JSON_DOC_FIELD_PATH).asString().getValue());

src/main/java/com/mongodb/kafka/connect/sink/cdc/debezium/mongodb/MongoDbUpdate.java

@@ -47,6 +47,10 @@ public WriteModel<BsonDocument> perform(final SinkDocument doc) {
             .orElseThrow(
                 () -> new DataException("Value document must not be missing for update operation"));
 
+    if (!valueDoc.containsKey(JSON_DOC_FIELD_PATH)) {
+      throw new DataException(format("Update document missing `%s` field.", JSON_DOC_FIELD_PATH));
+    }
+
     try {
       BsonDocument updateDoc =
           BsonDocument.parse(valueDoc.getString(JSON_DOC_FIELD_PATH).getValue());
@@ -69,6 +73,10 @@ public WriteModel<BsonDocument> perform(final SinkDocument doc) {
               .orElseThrow(
                   () -> new DataException("Key document must not be missing for update operation"));
 
+      if (!keyDoc.containsKey(JSON_ID_FIELD)) {
+        throw new DataException(format("Update document missing `%s` field.", JSON_ID_FIELD));
+      }
+
       BsonDocument filterDoc =
           BsonDocument.parse(
               format("{%s: %s}", ID_FIELD, keyDoc.getString(JSON_ID_FIELD).getValue()));