Merge pull request #76 from marklogic-community/feature/bulkds-addition

Tweaks to Bulk Data Services feature

Author: rjrudin

README.md

@@ -337,6 +337,17 @@ name starts with `ml.document.`. These properties are included to support an end
 dynamic endpoint that is driven by the values of these properties instead of them being hardcoded within the 
 endpoint module. This can allow for reusing the same endpoint module across multiple connector instances.
 
+Note that when using this approach, none of the properties starting with the following prefixes will have any impact, 
+as they are specific to the approach that uses DMSDK:
+
+- `ml.datahub`
+- `ml.dmsdk`
+- `ml.document`  
+- `ml.id.strategy`
+
+As noted above, the `ml.document` properties will be included in `endpointConstants`, but they otherwise have no impact
+because the endpoint developer chooses whether to apply them or not in their endpoint module.
+
 #### Configuring Bulk Data Services performance
 
 MarkLogic's Bulk Data Services feature is designed to leverage the multi-threading and parallelization support provided

config/marklogic-sink.properties

@@ -127,3 +127,11 @@ ml.dmsdk.threadCount=8
 # ml.datahub.flow.steps=
 # Set to true to log at the info level the response data from running a flow
 # ml.datahub.flow.logResponse=true
+
+# Defines the URI of a Bulk Data Services API declaration. Requires that ml.connection.modulesDatabase be set. See the
+# user guide for more information on using Bulk Data Services instead of DMSDK for writing data to MarkLogic.
+# ml.sink.bulkds.apiUri=
+
+# Required when using Bulk Data Services so that the API declaration can be retrieved from MarkLogic.
+# ml.connection.modulesDatabase=
+

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

@@ -67,9 +67,9 @@ protected void onStart(Map<String, Object> parsedConfig) {
     @Override
     public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
         if (bulkInputCaller != null) {
-            logger.info("Flushing BulkInputCaller");
+            logger.debug("Flushing BulkInputCaller");
             bulkInputCaller.awaitCompletion();
-            logger.info("Finished flushing BulkInputCaller");
+            logger.debug("Finished flushing BulkInputCaller");
         }
     }
 

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

@@ -94,6 +94,10 @@ public class MarkLogicSinkConfig extends AbstractConfig {
         .define(SSL_MUTUAL_AUTH, Type.BOOLEAN, null, Importance.LOW,
             "Set this to true for 2-way SSL; defaults to 1-way SSL")
 
+        .define(BULK_DS_API_URI, Type.STRING, null, Importance.LOW,
+            "Defines the URI of a Bulk Data Services API declaration. Requires that ml.connection.modulesDatabase be set. See the " +
+                "user guide for more information on using Bulk Data Services instead of DMSDK for writing data to MarkLogic.")
+
         .define(DOCUMENT_FORMAT, Type.STRING, null, Importance.MEDIUM,
             "Specify the format of each document; either 'JSON', 'XML', 'BINARY', 'TEXT', or 'UNKNOWN'. If not set, MarkLogic will determine the document type based on the ml.document.uriSuffix property.")
         .define(DOCUMENT_COLLECTIONS, Type.STRING, null, Importance.MEDIUM,
@@ -129,11 +133,6 @@ public class MarkLogicSinkConfig extends AbstractConfig {
         .define(DMSDK_INCLUDE_KAFKA_METADATA, Type.BOOLEAN, null, Importance.LOW,
             "Set to true so that Kafka record metadata is added to document metadata before it is written. If the document fails to be written, the Kafka record metadata will be logged as well.")
 
-        // TODO Need more info here on the API declaration itself?
-        .define(BULK_DS_API_URI, Type.STRING, null, Importance.LOW,
-            "Defines the URI of a Bulk Data Services API declaration. If set, all DMSDK properties will be ignored as Bulk Data Services will be used instead of DMSDK. " +
-                "Also, ml.connection.modulesDatabase must be defined so that the API declaration can be retrieved from the modules database.")
-
         .define(LOGGING_RECORD_KEY, Type.BOOLEAN, null, Importance.LOW,
             "Set to true to log at the info level the key of each record")
         .define(LOGGING_RECORD_HEADERS, Type.BOOLEAN, null, Importance.LOW,