Merge remote-tracking branch 'origin/dev-1.1.20' into dev-1.1.20

Author: yuankang134

dss-appconn/appconns/dss-eventchecker-appconn/src/main/java/com/webank/wedatasphere/dss/appconn/eventchecker/entity/EventChecker.java

@@ -45,7 +45,7 @@ public class EventChecker implements Runnable{
 	public final static String SAVE_KEY="msg.savekey";
 	public final static String USER_TIME="msg.init.querytime";
 	public final static String TODAY="only.receive.today";
-	public final static String USE_RUN_DATE ="use.rundate";
+	public final static String USE_RUN_DATE ="msg.receive.use.rundate";
 	public final static String AFTERSEND="msg.after.send";
 
 	private Properties p;

dss-appconn/appconns/dss-eventchecker-appconn/src/main/java/com/webank/wedatasphere/dss/appconn/eventchecker/service/AbstractEventCheckReceiver.java

@@ -138,7 +138,7 @@ String getOffset(int jobId, Properties props, Logger log){
      * Consistent entrance to consumer message
      */
     String[] getMsg(Properties props, Logger log,String ... params){
-        boolean useRunDate=Boolean.getBoolean(params[3]);
+        boolean useRunDate=Boolean.parseBoolean(params[3]);
         String sqlForReadTMsg;
         if(useRunDate){
             sqlForReadTMsg ="SELECT * FROM event_queue WHERE topic=? AND msg_name=? AND send_time >=? AND send_time <=? AND msg_id >?  AND run_date =?ORDER BY msg_id ASC LIMIT 1";
@@ -159,6 +159,7 @@ String[] getMsg(Properties props, Logger log,String ... params){
             pstmt.setString(4, params[1]);
             pstmt.setString(5, params[2]);
             if(useRunDate){
+                log.info("use run_date, run_date:{}", params[4]);
                 pstmt.setString(6,params[4]);
             }
             log.info("param {} StartTime: " + params[0] + ", EndTime: " + params[1]

dss-appconn/appconns/dss-eventchecker-appconn/src/main/java/com/webank/wedatasphere/dss/appconn/eventchecker/service/DefaultEventcheckReceiver.java

@@ -54,6 +54,7 @@ public boolean reciveMsg(int jobId, Properties props, Logger log) {
         try{
             String lastMsgId = getOffset(jobId,props,log);
             String[] executeType = createExecuteType(jobId,props,log,lastMsgId);
+            log.info("event receiver executeType[]:{},{},{},{},{}",executeType[0],executeType[1],executeType[2],executeType[3],executeType[4]);
             if(executeType!=null && executeType.length ==5){
                 String[] consumedMsgInfo = getMsg(props, log,executeType);
                 if(consumedMsgInfo!=null && consumedMsgInfo.length == 4){
@@ -77,17 +78,17 @@ private String[] createExecuteType(int jobId, Properties props, Logger log,Strin
         Boolean useRunDateFlag = receiveTodayFlag && (null != useRunDate && "true".equalsIgnoreCase(useRunDate.trim()));
         String[] executeType = null;
         try {
-            if(receiveTodayFlag){
-                if(afterSendFlag){
-                    executeType = new String[]{nowStartTime,todayEndTime,lastMsgId,useRunDateFlag.toString(),runDate};
-                }else{
-                    executeType = new String[]{todayStartTime,todayEndTime,lastMsgId,useRunDateFlag.toString(),runDate};
+            if (receiveTodayFlag && !useRunDateFlag) {
+                if (afterSendFlag) {
+                    executeType = new String[]{nowStartTime, todayEndTime, lastMsgId, useRunDateFlag.toString(), runDate};
+                } else {
+                    executeType = new String[]{todayStartTime, todayEndTime, lastMsgId, useRunDateFlag.toString(), runDate};
                 }
-            }else{
-                if(afterSendFlag){
-                    executeType = new String[]{nowStartTime,allEndTime,lastMsgId,useRunDateFlag.toString(),runDate};
-                }else{
-                    executeType = new String[]{allStartTime,allEndTime,lastMsgId,useRunDateFlag.toString(),runDate};
+            } else {
+                if (afterSendFlag) {
+                    executeType = new String[]{nowStartTime, allEndTime, lastMsgId, useRunDateFlag.toString(), runDate};
+                } else {
+                    executeType = new String[]{allStartTime, allEndTime, lastMsgId, useRunDateFlag.toString(), runDate};
                 }
             }
         }catch(Exception e){

dss-appconn/appconns/dss-eventchecker-appconn/src/main/scala/com/webank/wedatasphere/dss/appconn/eventchecker/execution/EventCheckerRefExecutionOperation.scala

@@ -76,10 +76,7 @@ class EventCheckerRefExecutionOperation
     val InstanceConfig = this.service.getAppInstance.getConfig
     val scalaParams: scala.collection.mutable.Map[String, Object] = requestRef.getExecutionRequestRefContext.getRuntimeMap
     val properties = new Properties()
-    val variableParams: scala.collection.mutable.Map[String, Object] = requestRef.getRefJobContent.get("variable").asInstanceOf[java.util.Map[String, Object]]
-    if (variableParams.exists(x => x._1.equalsIgnoreCase(VariableUtils.RUN_DATE))) {
-      properties.put(VariableUtils.RUN_DATE, variableParams.get(VariableUtils.RUN_DATE))
-    }
+    properties.put(VariableUtils.RUN_DATE,requestRef.getRunDate)
     InstanceConfig.foreach { record =>
       if(null == record._2) {
         properties.put(record._1, "")}