Repository: Vantiq/vantiq-extension-sources (language: python)

Issue:
Testing:  JDBC Async Processing tests flaky
Seems to be timing dependent connector config vs. starting to process messages.


Discussion / hints:

Fix patch:
diff --git a/build.gradle b/build.gradle
index 15d9954d..9acd36dd 100644
--- a/build.gradle
+++ b/build.gradle
@@ -39,7 +39,7 @@ ext {
     eclipseMiloVersion = '0.2.1'
     vantiqSDKVersion = '1.1.1'
     log4jVersion = '2.17.0'
-    lombokVersion = '1.18.24'
+    lombokVersion = '1.18.26'
     camelVersion = '3.18.3'
 }
 
diff --git a/jdbcSource/build.gradle b/jdbcSource/build.gradle
index 516bfbd4..d00b5058 100644
--- a/jdbcSource/build.gradle
+++ b/jdbcSource/build.gradle
@@ -3,6 +3,8 @@ version 'unspecified'
 
 apply plugin: 'java'
 apply plugin: 'application'
+apply plugin: 'io.franzbecker.gradle-lombok'
+
 
 sourceCompatibility = 1.8
 
@@ -57,6 +59,9 @@ tasks.withType(Test) {
 }
 
 dependencies {
+    annotationProcessor "org.projectlombok:lombok:${lombokVersion}"
+    implementation "org.projectlombok:lombok:${lombokVersion}"
+
     // This dependency is used internally, and not exposed to consumers on their own compile classpath.
     implementation 'com.google.guava:guava:23.0'
 
@@ -64,7 +69,8 @@ dependencies {
 
     implementation "org.slf4j:slf4j-api:1.7.25"
     implementation "org.apache.logging.log4j:log4j-slf4j-impl:${log4jVersion}"
-    
+    testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:${log4jVersion}"
+
     //  Compiling relevant JDBC Driver .jar file
     runtimeOnly files("${System.env.JDBC_DRIVER_LOC}")
     


Test patch:
diff --git a/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java b/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
index aab82ffe..abba2266 100644
--- a/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
+++ b/jdbcSource/src/test/java/io/vantiq/extsrc/jdbcSource/TestJDBC.java
@@ -37,12 +37,15 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import lombok.extern.slf4j.Slf4j;
+import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 @SuppressWarnings({"PMD.ExcessiveClassLength", "rawtypes", "PMD.IllegalTypeCheck"})
+@Slf4j
 public class TestJDBC extends TestJDBCBase {
     
     // Queries to be tested
@@ -203,6 +206,27 @@ public static void setupEnv() throws IOException {
         serverConfigFile.deleteOnExit();
         Utils.obtainServerConfig();
     }
+    
+    @After
+    public void postTestCleanup() {
+        // Delete the Source/Type/Topic/Procedure/Rule from VANTIQ
+        deleteType();
+        deleteTopic();
+        deleteProcedure();
+        deleteRule();
+        if (core != null) {
+            core.close();
+            core = null;
+        }
+        
+        if (jdbc != null) {
+            jdbc.close();
+            jdbc = null;
+        }
+        // Don't delete the source until you've shut down the connector. Otherwise, a bunch of spurious errors appear
+        // in the test log as it tries to reconnect.
+        deleteSource();
+    }
 
     @SuppressWarnings({"PMD.JUnit4TestShouldUseAfterAnnotation", "PMD.CognitiveComplexity", "PMD.EmptyCatchBlock"})
     @AfterClass
@@ -790,6 +814,7 @@ public void testDBReconnect() throws VantiqSQLException {
     }
     
     @Test
+//    @Ignore
     public void testMaxMessageSize() {
         // Only run test with intended vantiq availability
         assumeTrue(testAuthToken != null && testVantiqServer != null);
@@ -822,6 +847,8 @@ public void testMaxMessageSize() {
         VantiqResponse response = vantiq.query(testSourceName, params);
         JsonArray responseBody = (JsonArray) response.getBody();
         assert responseBody.size() == numRows;
+        assert core != null;
+        assert core.lastRowBundle != null;
         assert core.lastRowBundle.length == JDBCCore.DEFAULT_BUNDLE_SIZE;
 
         // Query with an invalid bundleFactor
@@ -884,15 +911,15 @@ public void testMaxMessageSize() {
 
     @Test
     public void testAsynchronousProcessing() throws InterruptedException {
-        doAsynchronousProcessing(true);
+        doAsynchronousProcessing(true, 500);
     }
 
     @Test
     public void testAsynchronousProcessingWithDefaults() throws InterruptedException {
-        doAsynchronousProcessing(false);
+        doAsynchronousProcessing(false, 300);
     }
 
-    public void doAsynchronousProcessing(boolean useCustomTaskConfig) throws InterruptedException {
+    public void doAsynchronousProcessing(boolean useCustomTaskConfig, int limitValue) throws InterruptedException {
         // Only run test with intended vantiq availability
         assumeTrue(testAuthToken != null && testVantiqServer != null);
         assumeTrue(testDBUsername != null && testDBPassword != null && testDBURL != null && jdbcDriverLoc != null);
@@ -906,7 +933,6 @@ public void doAsynchronousProcessing(boolean useCustomTaskConfig) throws Interru
 
         // Setup a VANTIQ JDBC Source, and start running the core
         setupSource(createSourceDef(true, useCustomTaskConfig));
-
         // Create Type to store query results
         setupAsynchType();
 
@@ -914,11 +940,16 @@ public void doAsynchronousProcessing(boolean useCustomTaskConfig) throws Interru
         setupTopic();
 
         // Create Procedure to publish to VANTIQ Source
-        setupProcedure();
+        setupProcedure(limitValue);
 
         // Create Rule to query the VANTIQ Source
         setupRule();
-
+    
+        // Select from the type and make sure all of our results are there as expected
+        VantiqResponse response = vantiq.select(testTypeName, null, null, null);
+        ArrayList responseBody = (ArrayList) response.getBody();
+        assertEquals (0, responseBody.size());
+    
         // Publish to the source in order to create a table
         Map<String, Object> createParams = new LinkedHashMap<>();
         createParams.put("query", CREATE_TABLE_ASYNCH);
@@ -926,15 +957,25 @@ public void doAsynchronousProcessing(boolean useCustomTaskConfig) throws Interru
 
 
         // Execute Procedure to trigger asynchronous publish/queries (assign to variable to ensure that procedure has finished before selecting from type)
-        VantiqResponse response = vantiq.execute(testProcedureName, new LinkedHashMap<>());
-
-        // Sleep for 10 seconds to allow all queries to finish
-        Thread.sleep(10000);
-
-        // Select from the type and make sure all of our results are there as expected
-        response = vantiq.select(testTypeName, null, null, null);
-        ArrayList responseBody = (ArrayList) response.getBody();
-        assertEquals (500, responseBody.size());
+        response = vantiq.execute(testProcedureName, new LinkedHashMap<>());
+        if (!response.isSuccess()) {
+            log.debug("Errors on execute {}: {}", testProcedureName, response.getErrors());
+        }
+        assert response.isSuccess();
+        // Sleep for a bit and see if we're done yet
+        long seconds = 1;
+        
+        while (seconds < 30) {
+            Thread.sleep(1000);
+            seconds += 1;
+            // Select from the type and make sure all of our results are there as expected
+            response = vantiq.select(testTypeName, null, null, null);
+            responseBody = (ArrayList) response.getBody();
+            if (limitValue == responseBody.size()) {
+                break;
+            }
+        }
+        assertEquals(limitValue, responseBody.size());
 
         // Delete the table for next test
         Map<String, Object> deleteParams = new LinkedHashMap<>();
@@ -1022,7 +1063,7 @@ public void testBatchProcessing() {
         // Creating a list of strings to insert as a batch
         ArrayList<String> batch = new ArrayList<>();
         for (int i = 0; i<50; i++) {
-            batch.add(INSERT_TABLE_BATCH);
+            batch.add("INSERT INTO TestBatch VALUES (" + i + ", 'First', 'Second');");
         }
 
         // Inserting data into the table as a batch
@@ -1035,6 +1076,11 @@ public void testBatchProcessing() {
         Map<String, Object> queryParams = new LinkedHashMap<>();
         queryParams.put("query", SELECT_TABLE_BATCH);
         response = vantiq.query(testSourceName, queryParams);
+        if (response.hasErrors()) {
+            log.debug("Response errors: {}", response.getErrors());
+        }
+        assert response.isSuccess();
+        assert response.getBody() instanceof JsonArray;
         JsonArray responseBody = (JsonArray) response.getBody();
         assert responseBody.size() == 50;
 
@@ -1051,43 +1097,61 @@ public void testQueryUpdate() {
         // Check that Source does not already exist in namespace, and skip test if they do
         assumeFalse(checkSourceExists());
 
-        // Setup a VANTIQ JDBC Source, and start running the core
-        setupSource(createSourceDef(false, false));
-
-        // Create table using query
-        Map<String, Object> createParams = new LinkedHashMap<>();
-        createParams.put("query", CREATE_TABLE_QUERY);
-        VantiqResponse response = vantiq.query(testSourceName, createParams);
-        assert !response.hasErrors();
-
-        // Inserting data into the table using query
-        Map<String, Object> insertParams = new LinkedHashMap<>();
-        insertParams.put("query", INSERT_TABLE_QUERY);
-        response = vantiq.query(testSourceName, insertParams);
-        assert !response.hasErrors();
-
-        // Select the data from table and make sure the previous queries worked
-        Map<String, Object> queryParams = new LinkedHashMap<>();
-        queryParams.put("query", SELECT_TABLE_QUERY);
-        response = vantiq.query(testSourceName, queryParams);
-        JsonArray responseBody = (JsonArray) response.getBody();
-        JsonObject bodyObject = responseBody.get(0).getAsJsonObject();
-        assert bodyObject.get("id").getAsInt() == 1;
-        assert bodyObject.get("name").getAsString().equals("Name");
-
-        // Delete data from table using query
-        Map<String, Object> deleteParams = new LinkedHashMap<>();
-        deleteParams.put("query", DELETE_ROWS_QUERY);
-        response = vantiq.query(testSourceName, deleteParams);
-        assert !response.hasErrors();
-
-        // Double check that the delete worked
-        response = vantiq.query(testSourceName, queryParams);
-        responseBody = (JsonArray) response.getBody();
-        assert responseBody.size() == 0;
+        try {
+            // Setup a VANTIQ JDBC Source, and start running the core
+            setupSource(createSourceDef(false, false));
+            
+            // Create table using query
+            Map<String, Object> createParams = new LinkedHashMap<>();
+            createParams.put("query", CREATE_TABLE_QUERY);
 
-        // Delete the Source
-        deleteSource();
+            VantiqResponse response = vantiq.query(testSourceName, createParams);
+            if (response.hasErrors()) {
+                log.debug("Errors on response to {}: {}", CREATE_TABLE_QUERY, response.getErrors());
+            }
+            assert !response.hasErrors();
+    
+            // Inserting data into the table using query
+            Map<String, Object> insertParams = new LinkedHashMap<>();
+            insertParams.put("query", INSERT_TABLE_QUERY);
+            response = vantiq.query(testSourceName, insertParams);
+            assert !response.hasErrors();
+    
+            // Select the data from table and make sure the previous queries worked
+            Map<String, Object> queryParams = new LinkedHashMap<>();
+            queryParams.put("query", SELECT_TABLE_QUERY);
+            response = vantiq.query(testSourceName, queryParams);
+            Object b = response.getBody();
+            JsonObject bodyObject;
+            JsonArray responseBody;
+            if (b instanceof JsonArray) {
+                log.debug("Got an array");
+                responseBody = (JsonArray) response.getBody();
+                bodyObject = responseBody.get(0).getAsJsonObject();
+            } else {
+                log.debug("Got object of class: {}", b.getClass().getName());
+                bodyObject = (JsonObject) b;
+            }
+            assert bodyObject.get("id").getAsInt() == 1;
+            assert bodyObject.get("name").getAsString().equals("Name");
+    
+            // Delete data from table using query
+            Map<String, Object> deleteParams = new LinkedHashMap<>();
+            deleteParams.put("query", DELETE_ROWS_QUERY);
+            response = vantiq.query(testSourceName, deleteParams);
+            assert !response.hasErrors();
+    
+            // Double check that the delete worked
+            response = vantiq.query(testSourceName, queryParams);
+            responseBody = (JsonArray) response.getBody();
+            assert responseBody == null || responseBody.size() == 0;
+        } catch (Exception e) {
+            log.error("Test failing with exception: ", e);
+            fail("Trapped exception: " + e.getMessage());
+        } finally {
+            // Delete the Source
+            deleteSource();
+        }
     }
 
     @Test
@@ -1098,42 +1162,43 @@ public void testQueryUpdateBatch() {
 
         // Check that Source does not already exist in namespace, and skip test if they do
         assumeFalse(checkSourceExists());
-
-        // Setup a VANTIQ JDBC Source, and start running the core
-        setupSource(createSourceDef(false, false));
-
-        // Create table
-        Map<String, Object> createParams = new LinkedHashMap<>();
-        createParams.put("query", CREATE_TABLE_BATCH_QUERY);
-        vantiq.query(testSourceName, createParams);
-
-        // Creating a list of strings to insert as a batch
-        ArrayList<String> batch = new ArrayList<>();
-        for (int i = 0; i<50; i++) {
-            batch.add(INSERT_TABLE_BATCH_QUERY);
+        try {
+            // Setup a VANTIQ JDBC Source, and start running the core
+            setupSource(createSourceDef(false, false));
+    
+            // Create table
+            Map<String, Object> createParams = new LinkedHashMap<>();
+            createParams.put("query", CREATE_TABLE_BATCH_QUERY);
+            vantiq.query(testSourceName, createParams);
+    
+            // Creating a list of strings to insert as a batch
+            ArrayList<String> batch = new ArrayList<>();
+            for (int i = 0; i < 50; i++) {
+                batch.add(INSERT_TABLE_BATCH_QUERY);
+            }
+    
+            // Inserting data into the table as a batch
+            Map<String, Object> insertParams = new LinkedHashMap<>();
+            insertParams.put("query", batch);
+            VantiqResponse response = vantiq.query(testSourceName, insertParams);
+            assert !response.hasErrors();
+    
+            // Select the data from table and make sure the response is valid
+            Map<String, Object> queryParams = new LinkedHashMap<>();
+            queryParams.put("query", SELECT_TABLE_BATCH_QUERY);
+            response = vantiq.query(testSourceName, queryParams);
+            JsonArray responseBody = (JsonArray) response.getBody();
+            assert responseBody.size() == 50;
+    
+            // Adding a select statement to batch, to make sure it does not execute
+            batch.add(SELECT_TABLE_BATCH_QUERY);
+            insertParams.put("query", batch);
+            response = vantiq.query(testSourceName, insertParams);
+            assert response.hasErrors();
+        } finally {
+            // Delete the Source
+            deleteSource();
         }
-
-        // Inserting data into the table as a batch
-        Map<String, Object> insertParams = new LinkedHashMap<>();
-        insertParams.put("query", batch);
-        VantiqResponse response = vantiq.query(testSourceName, insertParams);
-        assert !response.hasErrors();
-
-        // Select the data from table and make sure the response is valid
-        Map<String, Object> queryParams = new LinkedHashMap<>();
-        queryParams.put("query", SELECT_TABLE_BATCH_QUERY);
-        response = vantiq.query(testSourceName, queryParams);
-        JsonArray responseBody = (JsonArray) response.getBody();
-        assert responseBody.size() == 50;
-
-        // Adding a select statement to batch, to make sure it does not execute
-        batch.add(SELECT_TABLE_BATCH_QUERY);
-        insertParams.put("query", batch);
-        response = vantiq.query(testSourceName, insertParams);
-        assert response.hasErrors();
-
-        // Delete the Source
-        deleteSource();
     }
 
     // ================================================= Helper functions =================================================
@@ -1180,6 +1245,36 @@ public static void setupSource(Map<String, Object> sourceDef) {
             core = new JDBCCore(testSourceName, testAuthToken, testVantiqServer);
             core.start(CORE_START_TIMEOUT);
         }
+        if (insertResponse.hasErrors()) {
+            log.debug("Errors creating source: {}", insertResponse.getErrors());
+        }
+        assert insertResponse.isSuccess();
+    
+        // Now, wait for system to get started...
+        try {
+            int counter = 0;
+            while (!core.client.isAuthed() && counter++ < 50) {
+                Thread.sleep(500L);
+            }
+            assert core.client.isAuthed();
+            Map sourceConfig = (Map) sourceDef.get("config");
+            Map jdbcConfig = (Map) sourceConfig.get("jdbcConfig");
+            Map general = (Map) jdbcConfig.get("general");
+            Boolean isAsync = (Boolean) general.getOrDefault("asynchronousProcessing", Boolean.FALSE);
+            log.debug("Source {} is Async: {}", testSourceName, isAsync);
+            if (isAsync) {
+                counter = 0;
+                while (counter++ < 60) {
+                    if (core.publishPool == null || core.queryPool == null) {
+                        Thread.sleep(500L);
+                    }
+                }
+                assert core.publishPool != null;
+                assert core.queryPool != null;
+            }
+        } catch (InterruptedException e) {
+            fail("Interrupted while starting system: " + e.getMessage());
+        }
     }
     
     public static Map<String, Object> createSourceDef(boolean isAsynch, boolean useCustomTaskConfig) {
@@ -1200,7 +1295,7 @@ public static Map<String, Object> createSourceDef(boolean isAsynch, boolean useC
             general.put("asynchronousProcessing", true);
             if (useCustomTaskConfig) {
                 general.put("maxActiveTasks", 10);
-                general.put("maxQueuedTasks", 20);
+                general.put("maxQueuedTasks", 50);
             }
         }
         
@@ -1302,10 +1397,10 @@ public static boolean checkProcedureExists() {
         return !responseBody.isEmpty();
     }
 
-    public static void setupProcedure() {
+    public static void setupProcedure(int limitValue) {
         String procedure =
                 "PROCEDURE " + testProcedureName +  "()\n"
-                + "for (i in range(0,500)) {\n"
+                + "for (i in range(0," + limitValue + ")) {\n"
                     + "var sqlQuery = \"INSERT INTO TestAsynchProcessing VALUES(\" + i + \", 'FirstName', 'LastName')\"\n"
                     + "PUBLISH {query: sqlQuery} to SOURCE " + testSourceName + "\n"
                     + "PUBLISH {\"key\": i} TO TOPIC \"" + testTopicName + "\"\n"