fix bug of msg.receive.use.rundate

Author: xlinliu

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){

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, "")}