Merge branch 'dev-1.1.7-webank' of github.com:WeDataSphere/Exchangis …

…into dev-1.1.7-webank
WeDataSphere · Nov 18, 2024 · ab70605 · ab70605
2 parents 65e4908 + be46292
commit ab70605
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 5 deletions.
diff --git a/...a/com/webank/wedatasphere/exchangis/datasource/service/impl/DefaultDataSourceService.java b/...a/com/webank/wedatasphere/exchangis/datasource/service/impl/DefaultDataSourceService.java
@@ -859,7 +859,8 @@ public Map<String, Object> getDataSourceConnectParamsById(String operator, Long
                 "");
 
         Map<String, Object> connectParams = result.getConnectParams();
-        if (Objects.nonNull(connectParams) && StringUtils.equals(String.valueOf(connectParams.get("isEncrypt")), "1")) {
+        if (Objects.nonNull(connectParams) && connectParams.containsKey("password")
+                && StringUtils.equals(String.valueOf(connectParams.get("isEncrypt")), "1")) {
             String decrypt = AESUtils.decrypt(connectParams.get("password").toString(),
                     LINKIS_DATASOURCE_AES_KEY.getValue());
             connectParams.replace("password", decrypt);

diff --git a/exchangis-job/exchangis-job-launcher/pom.xml b/exchangis-job/exchangis-job-launcher/pom.xml
@@ -26,7 +26,7 @@
         <dependency>
             <groupId>org.apache.linkis</groupId>
             <artifactId>linkis-computation-client</artifactId>
-            <version>${linkis.version}</version>
+            <version>${linkis-computation-client.version}</version>
         </dependency>
 
     </dependencies>

diff --git a/...bank/wedatasphere/exchangis/job/server/execution/scheduler/tasks/SubmitSchedulerTask.java b/...bank/wedatasphere/exchangis/job/server/execution/scheduler/tasks/SubmitSchedulerTask.java
@@ -83,7 +83,7 @@ public SubmitSchedulerTask(RateLimitService rateLimitService, LaunchableExchangi
             try {
                 submitAble.set(submitCondition.call());
             } catch (Exception e) {
-                // Ignore
+                LOG.warn("Error to set submit condition from callback, cause by: ", e);
             }
         }
         // Set max retry

diff --git a/.../com/webank/wedatasphere/exchangis/job/server/execution/subscriber/NewInTaskObserver.java b/.../com/webank/wedatasphere/exchangis/job/server/execution/subscriber/NewInTaskObserver.java
@@ -86,10 +86,14 @@ public int subscribe(List<LaunchableExchangisTask> publishedTasks) throws Exchan
                                         }
                                     }
                                     return success;
+                                } else {
+                                    LOG.warn("Unable to get parallel, maybe reach the maxParallel");
                                 }
                                 noParallel.set(true);
                                 return false;
                     }, (submitTask, e) -> {
+                        LOG.info("Unable to launch the task with job_execution_id {} by user {}, decrease the parallel ,cause by {}"
+                                , launchableExchangisTask.getExecuteUser(), launchableExchangisTask.getJobExecutionId(), e);
                         TaskParallelRule parallelRule = parallelManager.getOrCreateRule(launchableExchangisTask.getExecuteUser(),
                                 TaskParallelManager.Operation.SUBMIT);
                         // Decrease the parallel
@@ -103,6 +107,7 @@ public int subscribe(List<LaunchableExchangisTask> publishedTasks) throws Exchan
                     if (submitSchedulerTask.isSubmitAble()) {
                         submitSchedulerTask.setTenancy(launchableExchangisTask.getExecuteUser());
                         try {
+                            LOG.info("Do submit task with id {} and job_execution_id {}", launchableExchangisTask.getId(), launchableExchangisTask.getJobExecutionId());
                             taskExecution.submit(submitSchedulerTask);
                         } catch (Exception e) {
                             // If the consumer queue is full?
@@ -144,8 +149,8 @@ public void discard(List<LaunchableExchangisTask> unsubscribeTasks) {
                     //Ignore
                 }
             }
-            Queue<LaunchableExchangisTask> queue = getCacheQueue();
-            unsubscribeTasks.forEach(queue::offer);
+//            Queue<LaunchableExchangisTask> queue = getCacheQueue();
+//            unsubscribeTasks.forEach(queue::offer);
         }
     }
 }
diff --git a/...ver/src/main/java/com/webank/wedatasphere/exchangis/job/server/utils/SQLCommandUtils.java b/...ver/src/main/java/com/webank/wedatasphere/exchangis/job/server/utils/SQLCommandUtils.java
@@ -37,7 +37,9 @@ public static String contactSql(List<?> tables, List<?> alias,
     private static String columnListSql(List<?> columns){
         StringBuilder builder = new StringBuilder();
         for(int i = 0; i < columns.size(); i++){
+            builder.append("`");
             builder.append(columns.get(i));
+            builder.append("`");
             if(i < columns.size() - 1){
                 builder.append(DEFAULT_COLUMN_SEPARATOR);
             }

diff --git a/pom.xml b/pom.xml
@@ -40,6 +40,7 @@
 		<revision>1.1.7-webank</revision>
 		<dss.version>1.8.0-SNAPSHOT</dss.version>
 		<linkis.version>1.5.0-wds-SNAPSHOT</linkis.version>
+		<linkis-computation-client.version>1.6.0-wds</linkis-computation-client.version>
 		<linkis.datasource.version>1.5.0-wds-SNAPSHOT</linkis.datasource.version>
 		<streamis.version>0.1.0-SNAPSHOT</streamis.version>
 		<maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version>