[Fix-363][Server] Fixed Multiple metric are configured for a Job, resulting in abnormal alarm information.

datavines-server/src/main/java/io/datavines/server/dqc/coordinator/operator/JobResultValidator.java

@@ -128,27 +128,27 @@ private void sendErrorEmail(Long jobExecutionId){
         DataSource dataSource = jobExternalService.getDataSourceService().getDataSourceById(dataSourceId);
         String dataSourceName = dataSource.getName();
         String dataSourceType = dataSource.getType();
-        JobExecutionResult jobExecutionResult = jobExternalService.getJobExecutionResultByJobExecutionId(jobExecution.getId());
+        List<JobExecutionResult> errorJobExecutionResultList = jobExternalService.listErrorJobExecutionResultByJobExecutionId(jobExecution.getId());
         boolean isEn = !LanguageUtils.isZhContext();
-        if (jobExecutionResult != null) {
-            MetricExecutionResult metricExecutionResult = new MetricExecutionResult();
-            BeanUtils.copyProperties(jobExecutionResult, metricExecutionResult);
-            List<String> messages = new ArrayList<>();
-            messages.add((isEn ? "Job Name : ": "作业名称: ") + jobName);
-            messages.add(String.format((isEn ? "Datasource : %s [%s] : ": "数据源 : %s [%s]: ") ,dataSourceType.toUpperCase(), dataSourceName));
-            String title = buildAlertSubject(metricExecutionResult, isEn);
-            String content = buildAlertMessage(messages, metricExecutionResult, jobExecution.getEngineType(), isEn);
-            message.setSubject(title);
-            message.setMessage(content);
-
-            saveIssue(jobId, title, content);
-
-            Map<SlaSenderMessage, Set<SlaConfigMessage>> config = slaNotificationService.getSlasNotificationConfigurationByJobId(jobId);
-            if (config.isEmpty()){
-                return;
+        if (CollectionUtils.isNotEmpty(errorJobExecutionResultList)) {
+            for(JobExecutionResult errorJobExecutionResult : errorJobExecutionResultList){
+                MetricExecutionResult metricExecutionResult = new MetricExecutionResult();
+                BeanUtils.copyProperties(errorJobExecutionResult, metricExecutionResult);
+                List<String> messages = new ArrayList<>();
+                messages.add((isEn ? "Job Name : ": "作业名称: ") + jobName);
+                messages.add(String.format((isEn ? "Datasource : %s [%s] : ": "数据源 : %s [%s]: ") ,dataSourceType.toUpperCase(), dataSourceName));
+                String title = buildAlertSubject(metricExecutionResult, isEn);
+                String content = buildAlertMessage(messages, metricExecutionResult, jobExecution.getEngineType(), isEn);
+                message.setSubject(title);
+                message.setMessage(content);
+                saveIssue(jobId, title, content);
+                Map<SlaSenderMessage, Set<SlaConfigMessage>> config = slaNotificationService.getSlasNotificationConfigurationByJobId(jobId);
+                if (config.isEmpty()){
+                    return;
+                }
+                notificationClient.notify(message, config);
             }
 
-            notificationClient.notify(message, config);
         }
     }
 

datavines-server/src/main/java/io/datavines/server/repository/service/JobExecutionResultService.java

@@ -34,6 +34,8 @@ public interface JobExecutionResultService extends IService<JobExecutionResult>
 
     JobExecutionResult getByJobExecutionId(long jobExecutionId);
 
+    List<JobExecutionResult> listByErrorJobExecutionId(long jobExecutionId);
+
     List<JobExecutionResult> listByJobExecutionId(long jobExecutionId);
 
     JobExecutionResultVO getResultVOByJobExecutionId(long jobExecutionId);

datavines-server/src/main/java/io/datavines/server/repository/service/impl/JobExecutionResultServiceImpl.java

@@ -89,6 +89,14 @@ public List<JobExecutionResult> listByJobExecutionId(long jobExecutionId) {
                 .orderByDesc("update_time"));
     }
 
+    @Override
+    public List<JobExecutionResult> listByErrorJobExecutionId(long jobExecutionId) {
+        return baseMapper.selectList(new QueryWrapper<JobExecutionResult>()
+                .eq("job_execution_id", jobExecutionId)
+                .eq("state", DqJobExecutionState.FAILURE.getCode())
+                .orderByDesc("update_time"));
+    }
+
     @Override
     public JobExecutionResultVO getResultVOByJobExecutionId(long jobExecutionId) {
 

datavines-server/src/main/java/io/datavines/server/repository/service/impl/JobExternalService.java

@@ -172,6 +172,10 @@ public JobExecutionResult getJobExecutionResultByJobExecutionId(long jobExecutio
         return jobExecutionResultService.getByJobExecutionId(jobExecutionId);
     }
 
+    public List<JobExecutionResult> listErrorJobExecutionResultByJobExecutionId(long jobExecutionId) {
+        return jobExecutionResultService.listByErrorJobExecutionId(jobExecutionId);
+    }
+
     public List<JobExecutionResult> listJobExecutionResultByJobExecutionId(long jobExecutionId) {
         return jobExecutionResultService.listByJobExecutionId(jobExecutionId);
     }