Merge pull request #155 from marklogic/feature/test-fix

Improving error message for when max value query fails

Jenkinsfile

@@ -14,7 +14,7 @@ pipeline{
   stages{
     stage('tests'){
       steps{
-        copyRPM 'Latest','11.0'
+        copyRPM 'Release','11.0.0'
         setUpML '$WORKSPACE/xdmp/src/Mark*.rpm'
         sh label:'setup', script: '''#!/bin/bash
         cd kafka-connector

src/main/java/com/marklogic/kafka/connect/source/QueryHandlerUtil.java

@@ -37,7 +37,17 @@ interface QueryHandlerUtil {
     static String executeMaxValuePlan(RowManager rowManager, PlanBuilder.Plan maxValuePlan, long serverTimestamp, String maxValueQuery) {
         JacksonHandle handle = new JacksonHandle();
         handle.setPointInTimeQueryTimestamp(serverTimestamp);
-        JacksonHandle result = rowManager.resultDoc(maxValuePlan, handle);
+
+        JacksonHandle result;
+        try {
+            result = rowManager.resultDoc(maxValuePlan, handle);
+        } catch (Exception e) {
+            throw new MarkLogicConnectorException(String.format(
+                "Unable to get max constraint value; query: %s; server timestamp: %d; cause: %s",
+                maxValueQuery, serverTimestamp, e.getMessage()
+            ), e);
+        }
+
         JsonNode valueNode = result.get().at("/rows/0/constraint/value");
         if (valueNode == null) {
             String message = String.format(

src/test/java/com/marklogic/kafka/connect/source/AbstractIntegrationSourceTest.java

@@ -41,6 +41,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -156,6 +157,7 @@ protected void verifyQueryReturnsExpectedRows(String constraintValue, int expect
         task.setConstraintValueStore(new TestConstraintValueStore((String)parsedConfig.get(MarkLogicSourceConfig.CONSTRAINT_COLUMN_NAME), constraintValue));
         try {
             List<SourceRecord> newRecords = task.poll();
+            assertNotNull(newRecords, "poll() unexpectedly returned null; params: " + params + "; constraintValue: " + constraintValue);
             assertEquals(expectedRowCount, newRecords.size());
             assertTrue(((String) newRecords.get(0).value()).contains(stringInFirstRecord),
                 "Did not find " + stringInFirstRecord + " in " + ((String) newRecords.get(0).value()));

src/test/java/com/marklogic/kafka/connect/source/QueryHandlerUtilTest.java

@@ -0,0 +1,28 @@
+package com.marklogic.kafka.connect.source;
+
+import com.marklogic.client.io.StringHandle;
+import com.marklogic.client.row.RawQueryDSLPlan;
+import com.marklogic.client.row.RowManager;
+import com.marklogic.kafka.connect.MarkLogicConnectorException;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QueryHandlerUtilTest extends AbstractIntegrationSourceTest {
+
+    @Test
+    void invalidMaxValueQuery() {
+        RowManager rowManager = getDatabaseClient().newRowManager();
+        final String invalidDsl = "This should fail";
+        RawQueryDSLPlan invalidQuery = rowManager.newRawQueryDSLPlan(new StringHandle(invalidDsl));
+
+        MarkLogicConnectorException ex = assertThrows(MarkLogicConnectorException.class,
+            () -> QueryHandlerUtil.executeMaxValuePlan(rowManager, invalidQuery, 0, invalidDsl));
+
+        assertTrue(ex.getMessage().startsWith("Unable to get max constraint value; query: This should fail"),
+            "If either our max value query has a bug in it, or the server has a bug in it with processing our max " +
+                "value query, an exception should be thrown that informs the user that the max value query failed " +
+                "along with what the query was");
+    }
+}