[feat] Add unit tests

Author: kenssa4eedfd

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/executor/LoadOptions.java

@@ -20,6 +20,7 @@
 import java.io.File;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
 
@@ -345,8 +346,8 @@ public final class LoadOptions implements Cloneable {
 
     @Parameter(names = {"--batch-failure-fallback"}, arity = 1,
                description = "Whether to fallback to single insert when batch insert fails. " +
-                             "Default: false")
-    public boolean batchFailureFallback = false;
+                             "Default: true")
+    public boolean batchFailureFallback = true;
 
     public String workModeString() {
         if (this.incrementalMode) {
@@ -425,27 +426,28 @@ public static LoadOptions parseOptions(String[] args) {
             options.maxParseErrors = Constants.NO_LIMIT;
             options.maxInsertErrors = Constants.NO_LIMIT;
         }
-        if (options.batchInsertThreads != CPUS) {
-            adjustConnectionPoolIfDefault(options);
+        if (Arrays.asList(args).contains("--parallel-count")) {
+            LOG.warn("Parameter --parallel-count is deprecated, " +
+                     "please use --parser-threads instead");
         }
+        adjustConnectionPoolIfDefault(options);
         return options;
     }
 
     private static void adjustConnectionPoolIfDefault(LoadOptions options) {
         int batchThreads = options.batchInsertThreads;
         int maxConn = options.maxConnections;
         int maxConnPerRoute = options.maxConnectionsPerRoute;
+
         if (maxConn == DEFAULT_MAX_CONNECTIONS && maxConn < batchThreads * 4) {
             options.maxConnections = batchThreads * 4;
-            LOG.info("Auto adjusted max-conn to {} based on " +
-                     "batch-insert-threads({})",
+            LOG.info("Auto adjusted max-conn to {} based on batch-insert-threads({})",
                      options.maxConnections, batchThreads);
         }
 
         if (maxConnPerRoute == DEFAULT_MAX_CONNECTIONS_PER_ROUTE && maxConnPerRoute < batchThreads * 2) {
             options.maxConnectionsPerRoute = batchThreads * 2;
-            LOG.info("Auto adjusted max-conn-per-route to {} based on " +
-                     "batch-insert-threads({})",
+            LOG.info("Auto adjusted max-conn-per-route to {} based on batch-insert-threads({})",
                      options.maxConnectionsPerRoute, batchThreads);
         }
     }

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/task/TaskManager.java

@@ -138,12 +138,13 @@ public void shutdown() {
 
     public void submitBatch(InputStruct struct, ElementMapping mapping,
                             List<Record> batch) {
-        if (this.context.stopped()) {
-            return;
-        }
         long start = System.currentTimeMillis();
         try {
             this.batchSemaphore.acquire();
+            if (this.context.stopped()) {
+                this.batchSemaphore.release();
+                return;
+            }
         } catch (InterruptedException e) {
             throw new LoadException("Interrupted while waiting to submit %s " +
                                     "batch in batch mode", e, mapping.type());
@@ -162,6 +163,7 @@ public void submitBatch(InputStruct struct, ElementMapping mapping,
                                  mapping.type(), e);
                         this.submitInSingle(struct, mapping, batch);
                     } else {
+                        summary.metrics(struct).minusFlighting(batch.size());
                         this.context.occurredError();
                         this.context.stopLoading();
                         Printer.printError("Batch insert %s failed, stop loading. Please check the logs",

hugegraph-loader/src/test/java/org/apache/hugegraph/loader/test/functional/FileLoadTest.java

@@ -2058,7 +2058,8 @@ public void testLoadIncrementalModeAndLoadFailure()
                 "-h", SERVER,
                 "--batch-insert-threads", "2",
                 "--max-parse-errors", "1",
-                "--test-mode", "false"
+                "--test-mode", "false",
+                "--parser-threads", "1"
         ));
         argsList.addAll(Arrays.asList("--username", "admin", "--password", "pa"));
 
@@ -2259,7 +2260,8 @@ public void testReloadJsonFailureFiles() throws IOException,
                 "-h", SERVER,
                 "--check-vertex", "true",
                 "--batch-insert-threads", "2",
-                "--test-mode", "false"
+                "--test-mode", "false",
+                "--parser-threads", "1"
         ));
         argsList.addAll(Arrays.asList("--username", "admin", "--password", "pa"));
         HugeGraphLoader loader = new HugeGraphLoader(argsList.toArray(new String[0]));
@@ -2564,7 +2566,8 @@ public void testSourceOrTargetPrimaryValueNull() {
                 "-g", GRAPH,
                 "-h", SERVER,
                 "--batch-insert-threads", "2",
-                "--test-mode", "true"
+                "--test-mode", "true",
+                "--parser-threads", "1"
         ));
 
         argsList.addAll(Arrays.asList("--username", "admin", "--password", "pa"));
@@ -3047,7 +3050,8 @@ public void testReadReachedMaxLines() {
                 "-h", SERVER,
                 "--max-read-lines", "4",
                 "--batch-insert-threads", "2",
-                "--test-mode", "true"
+                "--test-mode", "true",
+                "--parser-threads", "1"
         };
         loadWithAuth(args);