DHFPROD-7979: Allow ingest to write to DB other than STAGING or FINAL

Author: ryanjdew

marklogic-data-hub-api/src/main/java/com/marklogic/hub/HubClient.java

@@ -32,6 +32,8 @@ static HubClient withHubClientConfig(HubClientConfig hubClientConfig) {
 
     DatabaseClient getStagingClient();
 
+    DatabaseClient getStagingClient(String databaseName);
+
     DatabaseClient getFinalClient();
 
     DatabaseClient getJobsClient();

marklogic-data-hub-api/src/main/java/com/marklogic/hub/impl/HubClientImpl.java

@@ -25,6 +25,7 @@
 
 public class HubClientImpl implements HubClient {
 
+    private final HubClientConfig hubClientConfig;
     private String username;
     private DatabaseClient stagingClient;
     private DatabaseClient finalClient;
@@ -34,6 +35,7 @@ public class HubClientImpl implements HubClient {
     private ManageClient manageClient;
 
     public HubClientImpl(HubClientConfig hubClientConfig) {
+        this.hubClientConfig = hubClientConfig;
         username = hubClientConfig.getUsername();
         stagingClient = hubClientConfig.newStagingClient(null);
         finalClient = hubClientConfig.newFinalClient(null);
@@ -58,6 +60,11 @@ public DatabaseClient getStagingClient() {
         return stagingClient;
     }
 
+    @Override
+    public DatabaseClient getStagingClient(String databaseName) {
+        return hubClientConfig.newStagingClient(databaseName);
+    }
+
     @Override
     public DatabaseClient getFinalClient() {
         return finalClient;

marklogic-data-hub/src/main/java/com/marklogic/hub/step/impl/WriteStepRunner.java

@@ -447,9 +447,10 @@ private RunStepResponse runIngester(RunStepResponse runStepResponse, Collection<
 
         Vector<String> errorMessages = new Vector<>();
 
+        // Optimize for Hub databases associated with Hub app servers, but allow other values
         dataMovementManager = destinationDatabase.equals(hubClient.getDbName(DatabaseKind.FINAL)) ?
             hubClient.getFinalClient().newDataMovementManager() :
-            hubClient.getStagingClient().newDataMovementManager();
+            hubClient.getStagingClient(destinationDatabase).newDataMovementManager();
 
         HashMap<String, JobTicket> ticketWrapper = new HashMap<>();
 

marklogic-data-hub/src/test/java/com/marklogic/hub/flow/IngestToJobsTest.java

@@ -0,0 +1,66 @@
+package com.marklogic.hub.flow;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.marklogic.client.io.JacksonHandle;
+import com.marklogic.hub.AbstractHubCoreTest;
+import com.marklogic.hub.DatabaseKind;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class IngestToJobsTest extends AbstractHubCoreTest {
+    final String docUri = "/customers/customer1.json";
+    /**
+     * This is a test cover the scenario of ingesting to a database that isn't a Data Hub content DB (e.g., STAGING or FINAL).
+     * We use JOBS, since it is a database we know must exist.
+     */
+    @Test
+    void ingestToJobs() throws JsonProcessingException {
+        final String flowName = "ingestToFinal";
+
+        removeJobsDocument();
+        installReferenceModelProject();
+
+        copyArtifactsToJobs();
+
+        makeInputFilePathsAbsoluteInFlow(flowName);
+        FlowInputs flowInputs = new FlowInputs(flowName).withSteps("1");
+        // Using runtime options, change the target DB to JOBS
+        Map<String,Object> options = new HashMap<>();
+        flowInputs.withOption("targetDatabase", getHubConfig().getDbName(DatabaseKind.JOB));
+        RunFlowResponse runFlowResponse = runFlow(flowInputs);
+        assertEquals("finished", runFlowResponse.getJobStatus());
+        JsonNode rawDoc = getHubClient().getJobsClient().newJSONDocumentManager().read(docUri, new JacksonHandle()).get();
+        assertEquals("1", rawDoc.get("envelope").get("instance").get("CustomerID").asText(),
+                "Verifying that the single doc was ingested into the jobs database");
+    }
+
+    private void copyArtifactsToJobs() {
+        retryIfNecessary(() -> {
+            getClientByName(getHubClient().getDbName(DatabaseKind.JOB)).newServerEval().addVariable("source-db", getHubClient().getDbName(DatabaseKind.FINAL)).xquery(
+                    "declare variable $source-db external;\n" +
+                            "for $collection in ('http://marklogic.com/data-hub/flow','http://marklogic.com/data-hub/step-definition')" +
+                            "let $artifacts := xdmp:invoke-function(function() {\n" +
+                            "       map:new(\n" +
+                            "           fn:collection($collection) ! (map:entry(xdmp:node-uri(.), .))\n" +
+                            "       )\n" +
+                            "   }, map:entry('database', xdmp:database($source-db)))\n" +
+                            "for $uri in map:keys($artifacts)\n" +
+                            "return xdmp:document-insert($uri, map:get($artifacts, $uri), map:entry('collections', $collection))"
+            ).eval();
+        });
+    }
+
+    private void removeJobsDocument() {
+        try {
+            getClientByName(DatabaseKind.getName(DatabaseKind.JOB)).newDocumentManager().delete(docUri);
+        } catch (Exception e) {
+            // ignoring exception, since the document may not exist
+        }
+    }
+}