Merge pull request #78 from marklogic-community/feature/bulkds-improvements

Reworked Bulk DS approach to avoid reading from modules database

Author: rjrudin

README.md

@@ -164,8 +164,12 @@ recommended only for users with experience in writing and deploying custom code
 ### Writing data via configuration (DMSDK)
 
 The intent behind using DMSDK with the MarkLogic REST API is that as many aspects of writing data to MarkLogic can be 
-controlled via properties without the need to write any code. The following sections describe the various ways in 
-which this can be achieved.
+controlled via properties without the need to write any code.
+
+#### Security requirements
+
+The user that the connector authenticates as must have the `rest-reader` and `rest-writer` privileges in order to 
+write data via the MarkLogic REST API, which the connector depends upon.
 
 #### Configuring document URIs
 
@@ -268,13 +272,23 @@ typically a dataflow framework like Kafka that can support multiple workers writ
 MarkLogic Kafka connector utilizes Bulk Data Services to send Kafka record data to a custom endpoint in which a 
 developer can write any code they like to control how the data is processed. 
 
-Configuring the MarkLogic Kafka connector to use Bulk Data Services requires the following two properties:
+#### Security requirements
+
+Unlike when using the MarkLogic REST API, no specific privileges or roles are required in order for the connector to
+invoke a Bulk Data Services endpoint. Instead, the required privileges and/or roles for the MarkLogic user that the 
+connector authenticates as will be entirely determined by the Bulk Data Services endpoint implementation. 
 
-- `ml.sink.bulkds.apiUri` = the URI of the Bulk Data Services API declaration
-- `ml.connection.modulesDatabase` = the name of the modules database containing both the API declaration and the 
-  endpoint module that it refers to
+#### Configuring Bulk Data Services usage
 
-The MarkLogic Kafka connector expects the API declaration to be:
+Configuring the MarkLogic Kafka connector to use Bulk Data Services involves the following properties:
+
+- `ml.sink.bulkds.endpointUri` = the URI of the Bulk Data Services endpoint module
+- `ml.sink.bulkds.batchSize` = an optional batch size; defaults to 100. Note that if you include `$bulk/inputBatchSize` 
+in your API declaration, it will be ignored in favor of this property. 
+
+Bulk Data Services requires that your endpoint module have an API declaration. The URI of the API declaration must 
+match that of your endpoint, but with `.api` as a suffix instead of `.sjs` or `.xqy`. The MarkLogic Kafka connector 
+expects the API declaration to have the following configuration:
 
 ```
 {
@@ -292,15 +306,11 @@ The MarkLogic Kafka connector expects the API declaration to be:
       "multiple": true,
       "nullable": true
     }
-  ],
-  "$bulk": {
-    "inputBatchSize": 100
-  }
+  ]
 }
 ```
 
-The `endpoint` field must be the URI of the associated Bulk Data Services endpoint module. You are free to set the 
-`inputBatchSize` to any numeric value you want based on the expected workload for your connector. 
+The `endpoint` field should have the same value as the `ml.sink.bulkds.endpointUri` property.
 
 It is recommended to start your endpoint module implementation with the following code:
 
@@ -357,15 +367,12 @@ in parallel.
 
 A key design feature of Bulk Data Services to understand is that, unlike MarkLogic's Data Movement SDK, it does not 
 support asynchronous flushing of data. Bulk Data Services will not write any data to MarkLogic until it has a number of
-documents equalling that of the batch size in the API declaration. Importantly, partial batches will not be written 
-until either enough records are received to meet the batch size, or until Kafka invokes the "flush" operation on the
+documents equalling that of the `ml.sink.bulkds.batchSize` property. Importantly, partial batches will not be written 
+until either enough records are received to meet the batch size, or until Kafka invokes the `flush` operation on the
 MarkLogic Kafka connector. You can use the Kafka connector property named `offset.flush.interval` to control how often
 the flush operation is invoked. This is a synchronous operation, but you may wish to have this occur fairly regularly, 
 such as every 5 or 10 seconds, to ensure that partial batches of data are not waiting too long to be written to 
 MarkLogic.
 
-You may also want to configure the `inputBatchSize` field in your API declaration to see if increasing or decreasing
-this value results in greater performance. 
-
 As always with MarkLogic applications, use the [MarkLogic Monitoring dashboard](https://docs.marklogic.com/guide/monitoring/intro)
 to understand resource consumption and server performance while testing various connector settings. 

config/marklogic-sink.properties

@@ -128,10 +128,10 @@ ml.dmsdk.threadCount=8
 # 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=
+# Defines the URI of a Bulk Data Services endpoint for writing data.
+# See the user guide for more information on using Bulk Data Services instead of DMSDK for writing data to MarkLogic.
+# ml.sink.bulkds.endpointUri=
 
+# Sets the number of documents to be sent in a batch to the Bulk Data Services endpoint. The connector will not send any
+# documents to MarkLogic until it has a number matching this property or until Kafka invokes the 'flush' operation on the connector.
+# ml.sink.bulkds.batchSize=100

gradle.properties

@@ -22,4 +22,3 @@ mlAppName=kafka-test
 mlUsername=admin
 mlPassword=changeme-in-gradle-local.properties
 mlContentForestsPerHost=1
-mlModulePermissions=rest-extension-user,read,rest-extension-user,execute,rest-admin,update

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

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.dataservices.IOEndpoint;
@@ -43,7 +44,7 @@ protected void onStart(Map<String, Object> parsedConfig) {
         DatabaseClientConfig databaseClientConfig = new DefaultDatabaseClientConfigBuilder().buildDatabaseClientConfig(parsedConfig);
         this.databaseClient = new DefaultConfiguredDatabaseClientFactory().newDatabaseClient(databaseClientConfig);
 
-        JacksonHandle modulesHandle = readApiDeclarationFromMarkLogic(parsedConfig, databaseClientConfig);
+        JacksonHandle modulesHandle = new JacksonHandle(buildApiDeclaration(parsedConfig));
         InputCaller<JsonNode> inputCaller = InputCaller.on(databaseClient, modulesHandle, new JacksonHandle().withFormat(Format.JSON));
 
         IOEndpoint.CallContext callContext = inputCaller.newCallContext()
@@ -95,35 +96,35 @@ protected void writeSinkRecord(SinkRecord sinkRecord) {
     }
 
     /**
-     * Bulk Data Services requires that the API declaration exist in the modules database associated with the app
-     * server that the connector will talk to.
+     * Build an API declaration based on the user inputs for an endpoint URI and optional batch size. It's feasible to
+     * do this because the connector knows what the {@code params} array must be, and the documentation instructs the
+     * endpoint developer to use the same array of parameters. Building the API here also avoids having to read it
+     * from a modules database which requires either the xdmp-eval-in or xdbc-eval privilege.
      *
      * @param parsedConfig
-     * @param databaseClientConfig
      * @return
      */
-    private JacksonHandle readApiDeclarationFromMarkLogic(Map<String, Object> parsedConfig, DatabaseClientConfig databaseClientConfig) {
-        final String bulkApiUri = (String) parsedConfig.get(MarkLogicSinkConfig.BULK_DS_API_URI);
-        final String modulesDatabase = (String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE);
-        if (!StringUtils.hasText(modulesDatabase)) {
-            throw new IllegalArgumentException("Cannot read Bulk Data Services API declaration at URI: " + bulkApiUri
-                + "; no modules database configured. Please set the "
-                + MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE + " property.");
-        }
-
-        final String originalDatabase = databaseClientConfig.getDatabase();
-        try {
-            databaseClientConfig.setDatabase(modulesDatabase);
-            DatabaseClient modulesClient = new DefaultConfiguredDatabaseClientFactory().newDatabaseClient(databaseClientConfig);
-            return modulesClient.newJSONDocumentManager().read(bulkApiUri, new JacksonHandle().withFormat(Format.JSON));
-        } catch (Exception ex) {
-            // The stacktrace isn't of any value here for a user; the message below will provide sufficient information
-            // for debugging
-            throw new RuntimeException("Unable to read Bulk Data Services API declaration at URI: " + bulkApiUri +
-                "; modules database: " + modulesDatabase + "; cause: " + ex.getMessage());
-        } finally {
-            databaseClientConfig.setDatabase(originalDatabase);
+    private JsonNode buildApiDeclaration(Map<String, Object> parsedConfig) {
+        final String endpoint = parsedConfig.get(MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI).toString();
+
+        ObjectNode api = this.objectMapper.createObjectNode();
+        api.put("endpoint", endpoint);
+        ArrayNode params = api.putArray("params");
+        ObjectNode param = params.addObject();
+        param.put("name", "endpointConstants");
+        param.put("datatype", "jsonDocument");
+        param.put("multiple", false);
+        param.put("nullable", false);
+        param = params.addObject();
+        param.put("name", "input");
+        param.put("datatype", "jsonDocument");
+        param.put("multiple", true);
+        param.put("nullable", true);
+        if (parsedConfig.containsKey(MarkLogicSinkConfig.BULK_DS_BATCH_SIZE)) {
+            int batchSize = Integer.parseInt(parsedConfig.get(MarkLogicSinkConfig.BULK_DS_BATCH_SIZE).toString());
+            api.putObject("$bulk").put("inputBatchSize", batchSize);
         }
+        return api;
     }
 
     /**

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

@@ -15,7 +15,6 @@ public class MarkLogicSinkConfig extends AbstractConfig {
     public static final String CONNECTION_HOST = "ml.connection.host";
     public static final String CONNECTION_PORT = "ml.connection.port";
     public static final String CONNECTION_DATABASE = "ml.connection.database";
-    public static final String CONNECTION_MODULES_DATABASE = "ml.connection.modulesDatabase";
     public static final String CONNECTION_SECURITY_CONTEXT_TYPE = "ml.connection.securityContextType";
     public static final String CONNECTION_USERNAME = "ml.connection.username";
     public static final String CONNECTION_PASSWORD = "ml.connection.password";
@@ -36,7 +35,8 @@ public class MarkLogicSinkConfig extends AbstractConfig {
     public static final String DMSDK_TRANSFORM_PARAMS_DELIMITER = "ml.dmsdk.transformParamsDelimiter";
     public static final String DMSDK_INCLUDE_KAFKA_METADATA = "ml.dmsdk.includeKafkaMetadata";
 
-    public static final String BULK_DS_API_URI = "ml.sink.bulkds.apiUri";
+    public static final String BULK_DS_ENDPOINT_URI = "ml.sink.bulkds.endpointUri";
+    public static final String BULK_DS_BATCH_SIZE = "ml.sink.bulkds.batchSize";
 
     public static final String DOCUMENT_COLLECTIONS_ADD_TOPIC = "ml.document.addTopicToCollections";
     public static final String DOCUMENT_COLLECTIONS = "ml.document.collections";
@@ -77,8 +77,6 @@ public class MarkLogicSinkConfig extends AbstractConfig {
             "External name for 'KERBEROS' authentication")
         .define(CONNECTION_DATABASE, Type.STRING, null, Importance.LOW,
             "Name of a database to connect to. If your REST API server has a content database matching that of the one that you want to write documents to, you do not need to set this.")
-        .define(CONNECTION_MODULES_DATABASE, Type.STRING, null, Importance.MEDIUM,
-            "Name of the modules database associated with the app server; required if using Bulk Data Services so that the API module can be retrieved")
         .define(CONNECTION_TYPE, Type.STRING, null, Importance.MEDIUM,
             "Set to 'GATEWAY' when the host identified by ml.connection.host is a load balancer. See https://docs.marklogic.com/guide/java/data-movement#id_26583 for more information.")
         // Boolean fields must have a default value of null; otherwise, Confluent Platform, at least in version 7.2.1,
@@ -94,9 +92,12 @@ 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(BULK_DS_ENDPOINT_URI, Type.STRING, null, Importance.LOW,
+            "Defines the URI of a Bulk Data Services endpoint for writing data. " +
+                "See the user guide for more information on using Bulk Data Services instead of DMSDK for writing data to MarkLogic.")
+        .define(BULK_DS_BATCH_SIZE, Type.INT, 100, Importance.LOW,
+            "Sets the number of documents to be sent in a batch to the Bulk Data Services endpoint. The connector will not send any " +
+                "documents to MarkLogic until it has a number matching this property or until Kafka invokes the 'flush' operation on the connector.")
 
         .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.")

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

@@ -36,7 +36,7 @@ public void stop() {
 
     @Override
     public Class<? extends Task> taskClass() {
-        Class clazz = StringUtils.hasText(this.config.get(MarkLogicSinkConfig.BULK_DS_API_URI)) ?
+        Class clazz = StringUtils.hasText(this.config.get(MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI)) ?
             BulkDataServicesSinkTask.class :
             WriteBatcherSinkTask.class;
         LoggerFactory.getLogger(getClass()).info("Task class: " + clazz);

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

@@ -7,19 +7,20 @@
 
 import java.util.Arrays;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 public class WriteViaBulkDataServicesTest extends AbstractIntegrationTest {
 
     private final static String TEST_COLLECTION = "bulk-ds-test";
-    private final static String TEST_BULK_API_URI = "/example/bulk-endpoint.api";
-    private final static String MODULES_DATABASE = "kafka-test-modules";
+    private final static String TEST_BULK_ENDPOINT_URI = "/example/bulk-endpoint.sjs";
 
     @Test
     void twoBatches() {
         AbstractSinkTask task = startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, TEST_BULK_API_URI,
-            MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE, MODULES_DATABASE,
+            MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI, TEST_BULK_ENDPOINT_URI,
+            MarkLogicSinkConfig.BULK_DS_BATCH_SIZE, "3",
+
             // Turning on logging here to get coverage of this code; not possibly to verify the output via assertions
             MarkLogicSinkConfig.LOGGING_RECORD_HEADERS, "true",
             MarkLogicSinkConfig.LOGGING_RECORD_KEY, "true"
@@ -41,11 +42,29 @@ void twoBatches() {
         assertCollectionSize("All 4 records should have been written, as the flush call will TBD", TEST_COLLECTION, 4);
     }
 
+    @Test
+    void defaultBatchSize() {
+        AbstractSinkTask task = startSinkTask(
+            MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI, TEST_BULK_ENDPOINT_URI
+        );
+
+        for (int i = 1; i < 100; i++) {
+            task.put(Arrays.asList(newSinkRecord("<anything/>")));
+        }
+        assertCollectionSize("The default batch size is expected to be 100, so no docs should have been written since " +
+            "we've only put 99", TEST_COLLECTION, 0);
+
+        task.put(Arrays.asList(newSinkRecord("<anything/>")));
+        assertCollectionSize(TEST_COLLECTION, 100);
+
+        task.put(Arrays.asList(newSinkRecord("<anything/>")));
+        assertCollectionSize(TEST_COLLECTION, 100);
+    }
+
     @Test
     void verifyEndpointConstants() {
         AbstractSinkTask task = startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, TEST_BULK_API_URI,
-            MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE, MODULES_DATABASE,
+            MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI, TEST_BULK_ENDPOINT_URI,
             MarkLogicSinkConfig.DOCUMENT_COLLECTIONS, "json-test",
             MarkLogicSinkConfig.DOCUMENT_URI_PREFIX, "/json/",
             MarkLogicSinkConfig.DOCUMENT_URI_SUFFIX, ".json",
@@ -92,8 +111,7 @@ void verifyEndpointConstants() {
     @Test
     void verifyKafkaMetadata() {
         AbstractSinkTask task = startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, TEST_BULK_API_URI,
-            MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE, MODULES_DATABASE
+            MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI, TEST_BULK_ENDPOINT_URI
         );
 
         final String topic = "topic1";
@@ -119,32 +137,10 @@ void verifyKafkaMetadata() {
         assertEquals(timestamp, metadata.get("timestamp").asLong());
     }
 
-    @Test
-    void missingModulesDatabase() {
-        IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, () -> startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, TEST_BULK_API_URI
-        ));
-        assertTrue(ex.getMessage().contains("Cannot read Bulk Data Services API declaration"),
-            "Unexpected error: " + ex.getMessage());
-    }
-
-    @Test
-    void invalidApiUri() {
-        RuntimeException ex = assertThrows(RuntimeException.class, () -> startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, "/this-doesnt-exist.api",
-            MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE, MODULES_DATABASE
-        ));
-        String message = ex.getMessage();
-        assertTrue(message.startsWith("Unable to read Bulk Data Services API declaration at URI: /this-doesnt-exist.api"),
-            "Unexpected message: " + message);
-        assertTrue(message.contains("Could not read non-existent document"), "Unexpected message: " + message);
-    }
-
     @Test
     void badBulkEndpoint() {
         AbstractSinkTask task = startSinkTask(
-            MarkLogicSinkConfig.BULK_DS_API_URI, "/example/bad-bulk-endpoint.api",
-            MarkLogicSinkConfig.CONNECTION_MODULES_DATABASE, MODULES_DATABASE
+            MarkLogicSinkConfig.BULK_DS_ENDPOINT_URI, "/example/bad-bulk-endpoint.sjs"
         );
 
         putAndFlushRecords(task, newSinkRecord("content-doesnt-matter"));

src/test/ml-config/security/roles/minimal-user-role.json

@@ -1,9 +1,6 @@
 {
   "role-name": "kafka-test-minimal-user",
-  "description": "rest-reader/rest-writer privileges are needed to read forest info and write documents; rest-extension-user, any-uri, unprotected-collections, and xdmp-eval-in is needed for Bulk DS to read the API declaration",
-  "role": [
-    "rest-extension-user"
-  ],
+  "description": "rest-reader/rest-writer privileges are needed to read forest info and write documents; any-uri and unprotected-collections are needed by the test endpoint to insert a document",
   "privilege": [
     {
       "privilege-name": "rest-reader",
@@ -24,11 +21,6 @@
       "privilege-name": "unprotected-collections",
       "action": "http://marklogic.com/xdmp/privileges/unprotected-collections",
       "kind": "execute"
-    },
-    {
-      "privilege-name": "xdmp:eval-in",
-      "action": "http://marklogic.com/xdmp/privileges/xdmp-eval-in",
-      "kind": "execute"
     }
   ]
 }