(loader): enhanced and checked logics

Author: sadwitdastreetz

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/HugeGraphLoader.java

@@ -226,7 +226,7 @@ public boolean load() {
             throw e;
         }
 
-        return this.context.noError();
+        return true;
     }
 
     public void shutdown() {
@@ -278,7 +278,9 @@ private void createSchema() {
                                                     .ifNotExist()
                                                     .dataType(config.getIdFieldType())
                                                     .build();
-                    client.schema().addPropertyKey(propertyKey);
+                    if (client.schema().getPropertyKey(config.getIdFieldName()) == null) {
+                        client.schema().addPropertyKey(propertyKey);
+                    }
                 }
                 groovyExecutor.execute(script, client);
                 List<VertexLabel> vertexLabels = client.schema().getVertexLabels();
@@ -646,7 +648,7 @@ private List<InputTaskItem> prepareTaskItems(List<InputStruct> structs,
                         try {
                             r.close();
                         } catch (Exception ex) {
-                            LOG.warn("Failed to close reader: {}", ex.getMessage());
+                            LOG.warn("Failed to close reader", ex);
                         }
                     }
                 }
@@ -678,29 +680,27 @@ private void loadStructs(List<InputStruct> structs) {
                  this.context.options().endFile,
                  scatter ? "scatter" : "sequential");
 
-        ExecutorService loadService = ExecutorUtil.newFixedThreadPool(parallelCount,
-                                                                      "loader");
-
-        List<InputTaskItem> taskItems = prepareTaskItems(structs, scatter);
-        List<CompletableFuture<Void>> loadTasks = new ArrayList<>();
+        ExecutorService loadService = null;
+        try {
+            loadService = ExecutorUtil.newFixedThreadPool(parallelCount, "loader");
+            List<InputTaskItem> taskItems = prepareTaskItems(structs, scatter);
+            List<CompletableFuture<Void>> loadTasks = new ArrayList<>();
 
-        if (taskItems.isEmpty()) {
-            LOG.info("No tasks to execute after filtering");
-            return;
-        }
+            if (taskItems.isEmpty()) {
+                LOG.info("No tasks to execute after filtering");
+                return;
+            }
 
-        for (InputTaskItem item : taskItems) {
-            // Init reader
-            item.reader.init(this.context, item.struct);
-            // Load data from current input mapping
-            loadTasks.add(
-                    this.asyncLoadStruct(item.struct, item.reader,
-                                         loadService));
-        }
+            for (InputTaskItem item : taskItems) {
+                // Init reader
+                item.reader.init(this.context, item.struct);
+                // Load data from current input mapping
+                loadTasks.add(
+                        this.asyncLoadStruct(item.struct, item.reader,
+                                             loadService));
+            }
 
-        LOG.info("waiting for loading finish {}", loadTasks.size());
-        // wait for finish
-        try {
+            LOG.info("waiting for loading finish {}", loadTasks.size());
             CompletableFuture.allOf(loadTasks.toArray(new CompletableFuture[0]))
                              .join();
         } catch (CompletionException e) {
@@ -723,8 +723,10 @@ private void loadStructs(List<InputStruct> structs) {
         } finally {
             // Shutdown service
             cleanupEmptyProgress();
-            loadService.shutdown();
-            LOG.info("load end");
+            if (loadService != null) {
+                loadService.shutdownNow();
+            }
+            LOG.info("Load end");
         }
     }
 

hugegraph-loader/src/main/java/org/apache/hugegraph/loader/reader/jdbc/Fetcher.java

@@ -68,8 +68,6 @@ private Connection connect() throws SQLException {
                 password);
     }
 
-    ;
-
     abstract String[] readHeader() throws SQLException;
 
     abstract void readPrimaryKey() throws SQLException;

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

@@ -2033,15 +2033,15 @@ public void testLoadIncrementalModeAndLoadFailure()
            throws IOException, InterruptedException {
         ioUtil.write("vertex_person.csv",
                      "name,age,city",
-                     "marko，应该是数字，Beijing",
+                     "marko,应该是数字,Beijing",
                      "vadas,27,Hongkong",
                      "josh,32,Beijing",
                      "peter,35,Shanghai",
                      "\"li,nary\",26,\"Wu,han\"");
         ioUtil.write("vertex_software.csv", GBK,
                      "name,lang,price",
                      "office,C#,999",
-                     "lop,java，应该是数字",
+                     "lop,java,应该是数字",
                      "ripple,java,199");
 
         // 1st time
@@ -2096,7 +2096,7 @@ public void testLoadIncrementalModeAndLoadFailure()
         List<String> personFailureLines = FileUtils.readLines(personFailureFile,
                                                               Constants.CHARSET);
         Assert.assertEquals(2, personFailureLines.size());
-        Assert.assertEquals("marko，应该是数字，Beijing",
+        Assert.assertEquals("marko,应该是数字,Beijing",
                             personFailureLines.get(1));
 
         // 2nd time, incremental-mode
@@ -2159,14 +2159,14 @@ public void testLoadIncrementalModeAndLoadFailure()
         personFailureLines = FileUtils.readLines(personFailureFile,
                                                  Constants.CHARSET);
         Assert.assertEquals(2, personFailureLines.size());
-        Assert.assertEquals("marko，应该是数字，Beijing",
+        Assert.assertEquals("marko,应该是数字,Beijing",
                             personFailureLines.get(1));
 
         File softwareFailureFile = files[2];
         List<String> softwareFailureLines = FileUtils.readLines(
                                             softwareFailureFile, GBK);
         Assert.assertEquals(2, softwareFailureLines.size());
-        Assert.assertEquals("lop,java，应该是数字", softwareFailureLines.get(1));
+        Assert.assertEquals("lop,java,应该是数字", softwareFailureLines.get(1));
 
         // TODO: Change only one line first, and make the second line go wrong
         // modify person and software failure file