Improve "Could not convert value ..." messages (#140)

A viewer should not be left guessing whether the value was a null reference, or a "null" `CharSequence`.

Author: stIncMale

src/integrationTest/java/com/mongodb/kafka/connect/sink/MongoSinkTaskIntegrationTest.java

@@ -338,7 +338,9 @@ void testSinkCanHandleInvalidDocumentWhenErrorToleranceIsAll() {
       task.start(cfg);
 
       DataException e = assertThrows(DataException.class, () -> task.put(sinkRecords));
-      assertTrue(e.getMessage().contains("Could not convert value `a` into a BsonDocument"));
+      assertTrue(
+          e.getMessage()
+              .contains("Could not convert value 'a' (java.lang.String) into a BsonDocument"));
     }
   }
 

src/main/java/com/mongodb/kafka/connect/sink/converter/LazyBsonDocument.java

@@ -32,6 +32,8 @@
 import org.bson.BsonDocument;
 import org.bson.BsonValue;
 
+import com.mongodb.lang.Nullable;
+
 public class LazyBsonDocument extends BsonDocument {
   private static final long serialVersionUID = 1L;
 
@@ -158,15 +160,21 @@ private BsonDocument getUnwrapped() {
             unwrapped = converter.apply(record.keySchema(), record.key());
           } catch (Exception e) {
             throw new DataException(
-                format("Could not convert key `%s` into a BsonDocument.", record.key()), e);
+                format(
+                    "Could not convert key %s into a BsonDocument.",
+                    unambiguousToString(record.key())),
+                e);
           }
           break;
         case VALUE:
           try {
             unwrapped = converter.apply(record.valueSchema(), record.value());
           } catch (Exception e) {
             throw new DataException(
-                format("Could not convert value `%s` into a BsonDocument.", record.value()), e);
+                format(
+                    "Could not convert value %s into a BsonDocument.",
+                    unambiguousToString(record.value())),
+                e);
           }
           break;
         default:
@@ -185,4 +193,15 @@ private Object writeReplace() {
   private void readObject(final ObjectInputStream stream) throws InvalidObjectException {
     throw new InvalidObjectException("Proxy required");
   }
+
+  private static String unambiguousToString(@Nullable final Object v) {
+    String vToString = String.valueOf(v);
+    if (v == null) {
+      return format("'%s' (null reference)", vToString);
+    } else if (vToString.equals(String.valueOf((Object) null))) {
+      return format("'%s' (%s, not a null reference)", vToString, v.getClass().getName());
+    } else {
+      return format("'%s' (%s)", vToString, v.getClass().getName());
+    }
+  }
 }