Feature/kst 8 null objects (#15)

* kst-8 - First pass at fixing the bug with handling messages with a null value. These can be "tombstone" messages. Add 2 tests. One test for a blank message (""), and one for a null value.

* kst-8 - First pass at fixing the bug with handling messages with a null value. These can be "tombstone" messages. Add 2 tests. One test for a blank message (""), and one for a null value.

* kst-8 - Implementing review recommendations.
Throwing NullPointerExceptions in lower level classes.
Logging null records and values in the MarkLogicSinkTask to catch those early in the process.

* kst-8 - First pass at fixing the bug with handling messages with a null value. These can be "tombstone" messages. Add 2 tests. One test for a blank message (""), and one for a null value.

* kst-8 - Implementing review recommendations.
Throwing NullPointerExceptions in lower level classes.
Logging null records and values in the MarkLogicSinkTask to catch those early in the process.

* fixing a typo

Author: BillFarber

README.md

@@ -78,7 +78,7 @@ When a document is received and written by the connector, you'll see logging lik
 | ml.dmsdk.batchSize | 100 | Sets the number of documents to be written in a batch to MarkLogic. This may not have any impact depending on the connector receives data from Kafka, as the connector calls flushAsync on the DMSDK WriteBatcher after processing every collection of records. Thus, if the connector never receives at one time more than the value of this property, then the value of this property will have no impact. |
 | ml.dmsdk.threadCount | 8 | Sets the number of threads used by the Data Movement SDK for parallelizing writes to MarkLogic. Similar to the batch size property above, this may never come into play depending on how many records the connector receives at once. |
 | ml.document.collections | kafka-data | Optional - a comma-separated list of collections that each document should be written to |
-|ml.document.addTopicToCollections | false | Sets this to true so that the name of the topic that the connector reads from is added as a collection to each document inserted by the connector
+| ml.document.addTopicToCollections | false | Set this to true so that the name of the topic that the connector reads from is added as a collection to each document inserted by the connector
 | ml.document.format | JSON | Optional - specify the format of each document; either JSON, XML, BINARY, TEXT, or UNKNOWN |
 | ml.document.mimeType | (empty) | Optional - specify a mime type for each document; typically the format property above will be used instead of this |
 | ml.document.permissions | rest-reader,read,rest-writer,update | Optional - a comma-separated list of roles and capabilities that define the permissions for each document written to MarkLogic |

src/main/java/com/marklogic/client/ext/document/DocumentWriteOperationBuilder.java

@@ -17,6 +17,10 @@ public class DocumentWriteOperationBuilder {
 	private ContentIdExtractor contentIdExtractor = new DefaultContentIdExtractor();
 
 	public DocumentWriteOperation build(AbstractWriteHandle content, DocumentMetadataHandle metadata ) {
+		if (content == null) {
+			throw new NullPointerException("'content' must not be null");
+		}
+
 		if (hasText(collections)) {
 			metadata.getCollections().addAll(collections.trim().split(","));
 		}

src/main/java/com/marklogic/kafka/connect/sink/DefaultSinkRecordConverter.java

@@ -69,6 +69,9 @@ protected DocumentMetadataHandle addTopicToCollections (String topic, Boolean ad
 	 * @return
 	 */
 	protected AbstractWriteHandle toContent(SinkRecord record) {
+		if ((record == null) || (record.value() == null)) {
+			throw new NullPointerException("'record' must not be null, and must have a value.");
+		}
 		Object value = record.value();
 		if (value instanceof byte[]) {
 			BytesHandle content = new BytesHandle((byte[]) value);

src/main/java/com/marklogic/kafka/connect/sink/MarkLogicSinkTask.java

@@ -111,13 +111,25 @@ public void put(final Collection<SinkRecord> records) {
 		}
 
 		records.forEach(record -> {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Processing record value {} in topic {}", record.value(), record.topic());
+			if (record == null) {
+				logger.warn("Skipping null record object.");
+			} else {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Processing record value {} in topic {}", record.value(), record.topic());
+				}
+				if (record.value() != null) {
+					writeBatcher.add(sinkRecordConverter.convert(record));
+				} else {
+					logger.warn("Skipping record with null value - possibly a 'tombstone' message.");
+				}
 			}
-			writeBatcher.add(sinkRecordConverter.convert(record));
 		});
 
-		writeBatcher.flushAsync();
+		if (writeBatcher != null) {
+			writeBatcher.flushAsync();
+		} else {
+			logger.warn("writeBatcher is null - ignore this is you are running unit tests, otherwise this is a problem.");
+		}
 	}
 
 	public String version() {

src/test/java/com/marklogic/kafka/connect/sink/ConvertSinkRecordTest.java

@@ -8,9 +8,7 @@
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.junit.jupiter.api.Test;
 
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
+import java.util.*;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -21,6 +19,7 @@
 public class ConvertSinkRecordTest {
 
 	DefaultSinkRecordConverter converter;
+	MarkLogicSinkTask markLogicSinkTask = new MarkLogicSinkTask();
 
 	@Test
 	public void allPropertiesSet() {
@@ -78,6 +77,20 @@ public void binaryContent() {
 		assertEquals("hello world".getBytes().length, content.get().length);
 	}
 
+	@Test
+	public void emptyContent() {
+		final Collection<SinkRecord> records = new ArrayList<>();
+		records.add(newSinkRecord(null));
+		markLogicSinkTask.put(records);
+	}
+
+	@Test
+	public void nullContent() {
+		final Collection<SinkRecord> records = new ArrayList<>();
+		records.add(null);
+		markLogicSinkTask.put(records);
+	}
+
 	private SinkRecord newSinkRecord(Object value) {
 		return new SinkRecord("test-topic", 1, null, null, null, value, 0);
 	}