Added key to SourceRecord

KAFKA-49

Author: rozza

src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java

@@ -133,7 +133,9 @@ public List<SourceRecord> poll() {
 
                 jsonDocument.ifPresent((json) -> {
                     LOGGER.trace("Adding {} to {}", json, topicName);
-                    sourceRecords.add(new SourceRecord(partition, sourceOffset, topicName, Schema.STRING_SCHEMA, json));
+                    String keyJson = new BsonDocument("_id", changeStreamDocument.get("_id")).toJson();
+                    sourceRecords.add(new SourceRecord(partition, sourceOffset, topicName, Schema.STRING_SCHEMA, keyJson,
+                            Schema.STRING_SCHEMA, json));
                 });
 
                 // If the cursor is invalidated add the record and return calls