Emailing Task and Test Notification (#8626)

* Emailing Task Notification [WIP]

* Emailing Test Result to Owners of table in case it is enabled

openmetadata-service/src/main/java/org/openmetadata/service/events/ChangeEventHandler.java

@@ -17,12 +17,8 @@ import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
 import static org.openmetadata.schema.type.EventType.ENTITY_DELETED;
 import static org.openmetadata.schema.type.EventType.ENTITY_SOFT_DELETED;
 import static org.openmetadata.schema.type.EventType.ENTITY_UPDATED;
-import static org.openmetadata.service.Entity.TEAM;
-import static org.openmetadata.service.Entity.USER;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -36,27 +32,21 @@ import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
 import org.openmetadata.schema.EntityInterface;
 import org.openmetadata.schema.entity.feed.Thread;
-import org.openmetadata.schema.entity.teams.Team;
-import org.openmetadata.schema.entity.teams.User;
-import org.openmetadata.schema.type.AnnouncementDetails;
 import org.openmetadata.schema.type.ChangeDescription;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.EntityReference;
 import org.openmetadata.schema.type.EventType;
-import org.openmetadata.schema.type.Post;
-import org.openmetadata.schema.type.Relationship;
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.OpenMetadataApplicationConfig;
 import org.openmetadata.service.filter.FilterRegistry;
 import org.openmetadata.service.jdbi3.CollectionDAO;
-import org.openmetadata.service.jdbi3.CollectionDAO.EntityRelationshipRecord;
 import org.openmetadata.service.jdbi3.FeedRepository;
-import org.openmetadata.service.resources.feeds.MessageParser;
 import org.openmetadata.service.resources.feeds.MessageParser.EntityLink;
 import org.openmetadata.service.socket.WebSocketManager;
 import org.openmetadata.service.util.ChangeEventParser;
 import org.openmetadata.service.util.FilterUtil;
 import org.openmetadata.service.util.JsonUtils;
+import org.openmetadata.service.util.NotificationHandler;
 import org.openmetadata.service.util.RestUtil;
 
 @Slf4j
@@ -64,11 +54,13 @@ public class ChangeEventHandler implements EventHandler {
   private CollectionDAO dao;
   private FeedRepository feedDao;
   private ObjectMapper mapper;
+  private NotificationHandler notificationHandler;
 
   public void init(OpenMetadataApplicationConfig config, Jdbi jdbi) {
     this.dao = jdbi.onDemand(CollectionDAO.class);
     this.feedDao = new FeedRepository(dao);
     this.mapper = new ObjectMapper();
+    this.notificationHandler = new NotificationHandler(jdbi.onDemand(CollectionDAO.class));
   }
 
   public Void process(ContainerRequestContext requestContext, ContainerResponseContext responseContext) {
@@ -76,7 +68,7 @@ public class ChangeEventHandler implements EventHandler {
     SecurityContext securityContext = requestContext.getSecurityContext();
     String loggedInUserName = securityContext.getUserPrincipal().getName();
     try {
-      handleWebSocket(responseContext);
+      notificationHandler.processNotifications(responseContext);
       ChangeEvent changeEvent = getChangeEvent(method, responseContext);
       if (changeEvent == null) {
         return null;
@@ -134,75 +126,6 @@ public class ChangeEventHandler implements EventHandler {
     return null;
   }
 
-  private void handleWebSocket(ContainerResponseContext responseContext) {
-    int responseCode = responseContext.getStatus();
-    if (responseCode == Status.CREATED.getStatusCode()
-        && responseContext.getEntity() != null
-        && responseContext.getEntity().getClass().equals(Thread.class)) {
-      Thread thread = (Thread) responseContext.getEntity();
-      try {
-        String jsonThread = mapper.writeValueAsString(thread);
-        switch (thread.getType()) {
-          case Task:
-            if (thread.getPostsCount() == 0) {
-              List<EntityReference> assignees = thread.getTask().getAssignees();
-              assignees.forEach(
-                  e -> {
-                    if (Entity.USER.equals(e.getType())) {
-                      WebSocketManager.getInstance()
-                          .sendToOne(e.getId(), WebSocketManager.TASK_BROADCAST_CHANNEL, jsonThread);
-                    } else if (Entity.TEAM.equals(e.getType())) {
-                      // fetch all that are there in the team
-                      List<EntityRelationshipRecord> records =
-                          dao.relationshipDAO()
-                              .findTo(e.getId().toString(), TEAM, Relationship.HAS.ordinal(), Entity.USER);
-                      WebSocketManager.getInstance()
-                          .sendToManyWithString(records, WebSocketManager.TASK_BROADCAST_CHANNEL, jsonThread);
-                    }
-                  });
-            }
-            break;
-          case Conversation:
-            WebSocketManager.getInstance().broadCastMessageToAll(WebSocketManager.FEED_BROADCAST_CHANNEL, jsonThread);
-            List<EntityLink> mentions;
-            if (thread.getPostsCount() == 0) {
-              mentions = MessageParser.getEntityLinks(thread.getMessage());
-            } else {
-              Post latestPost = thread.getPosts().get(thread.getPostsCount() - 1);
-              mentions = MessageParser.getEntityLinks(latestPost.getMessage());
-            }
-            mentions.forEach(
-                entityLink -> {
-                  String fqn = entityLink.getEntityFQN();
-                  if (USER.equals(entityLink.getEntityType())) {
-                    User user = dao.userDAO().findEntityByName(fqn);
-                    WebSocketManager.getInstance()
-                        .sendToOne(user.getId(), WebSocketManager.MENTION_CHANNEL, jsonThread);
-                  } else if (TEAM.equals(entityLink.getEntityType())) {
-                    Team team = dao.teamDAO().findEntityByName(fqn);
-                    // fetch all that are there in the team
-                    List<EntityRelationshipRecord> records =
-                        dao.relationshipDAO().findTo(team.getId().toString(), TEAM, Relationship.HAS.ordinal(), USER);
-                    WebSocketManager.getInstance()
-                        .sendToManyWithString(records, WebSocketManager.MENTION_CHANNEL, jsonThread);
-                  }
-                });
-            break;
-          case Announcement:
-            AnnouncementDetails announcementDetails = thread.getAnnouncement();
-            Long currentTimestamp = Instant.now().getEpochSecond();
-            if (announcementDetails.getStartTime() <= currentTimestamp
-                && currentTimestamp <= announcementDetails.getEndTime()) {
-              WebSocketManager.getInstance().broadCastMessageToAll(WebSocketManager.ANNOUNCEMENT_CHANNEL, jsonThread);
-            }
-            break;
-        }
-      } catch (JsonProcessingException e) {
-        throw new RuntimeException(e);
-      }
-    }
-  }
-
   public ChangeEvent getChangeEvent(String method, ContainerResponseContext responseContext) {
     // GET operations don't produce change events
     if (method.equals("GET")) {

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java

@@ -42,6 +42,8 @@ import org.jdbi.v3.sqlobject.customizer.BindMap;
 import org.jdbi.v3.sqlobject.customizer.Define;
 import org.jdbi.v3.sqlobject.statement.SqlQuery;
 import org.jdbi.v3.sqlobject.statement.SqlUpdate;
+import org.openmetadata.api.configuration.airflow.TaskNotificationConfiguration;
+import org.openmetadata.api.configuration.airflow.TestResultNotificationConfiguration;
 import org.openmetadata.common.utils.CommonUtil;
 import org.openmetadata.schema.TokenInterface;
 import org.openmetadata.schema.analytics.WebAnalyticEvent;
@@ -3164,9 +3166,17 @@ public interface CollectionDAO {
       settings.setConfigType(configType);
       Object value;
       try {
-        if (configType == SettingsType.ACTIVITY_FEED_FILTER_SETTING) {
+        switch (configType) {
+          case ACTIVITY_FEED_FILTER_SETTING:
             value = JsonUtils.readValue(json, new TypeReference<ArrayList<EventFilter>>() {});
-        } else {
+            break;
+          case TASK_NOTIFICATION_CONFIGURATION:
+            value = JsonUtils.readValue(json, TaskNotificationConfiguration.class);
+            break;
+          case TEST_RESULT_NOTIFICATION_CONFIGURATION:
+            value = JsonUtils.readValue(json, TestResultNotificationConfiguration.class);
+            break;
+          default:
             throw new RuntimeException("Invalid Settings Type");
         }
       } catch (IOException e) {

openmetadata-service/src/main/java/org/openmetadata/service/resources/settings/SettingsCache.java

@@ -24,9 +24,11 @@ import java.util.concurrent.TimeUnit;
 import javax.annotation.CheckForNull;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.settings.Settings;
+import org.openmetadata.schema.settings.SettingsType;
 import org.openmetadata.service.exception.EntityNotFoundException;
 import org.openmetadata.service.jdbi3.CollectionDAO;
 import org.openmetadata.service.jdbi3.SettingsRepository;
+import org.openmetadata.service.util.JsonUtils;
 
 @Slf4j
 public class SettingsCache {
@@ -60,6 +62,15 @@ public class SettingsCache {
     }
   }
 
+  public <T> T getSetting(SettingsType settingName, Class<T> clazz) throws RuntimeException {
+    try {
+      String json = JsonUtils.pojoToJson(SETTINGS_CACHE.get(settingName.toString()).getConfigValue());
+      return JsonUtils.readValue(json, clazz);
+    } catch (Exception ex) {
+      throw new RuntimeException(ex);
+    }
+  }
+
   public void putSettings(Settings setting) throws RuntimeException {
     SETTINGS_CACHE.put(setting.getConfigType().toString(), setting);
   }

openmetadata-service/src/main/java/org/openmetadata/service/socket/WebSocketManager.java

@@ -102,7 +102,7 @@ public class WebSocketManager {
     }
   }
 
-  public void sendToManyWithUUID(List<UUID> receivers, String event, String message) {
+  public void sendToManyWithUUID(HashSet<UUID> receivers, String event, String message) {
     receivers.forEach(e -> sendToOne(e, event, message));
   }
 

openmetadata-service/src/main/java/org/openmetadata/service/util/EmailUtil.java

@@ -15,7 +15,9 @@ import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.email.EmailRequest;
 import org.openmetadata.schema.email.SmtpSettings;
+import org.openmetadata.schema.entity.feed.Thread;
 import org.openmetadata.schema.entity.teams.User;
+import org.openmetadata.schema.tests.type.TestCaseResult;
 import org.simplejavamail.api.email.Email;
 import org.simplejavamail.api.email.EmailPopulatingBuilder;
 import org.simplejavamail.api.mailer.Mailer;
@@ -47,12 +49,13 @@ public class EmailUtil {
   public static final String ACTION_KEY = "action";
   public static final String ACTION_STATUS_KEY = "actionStatus";
   public static final String ACCOUNT_STATUS_TEMPLATE_FILE = "account-activity-change.ftl";
-
   private static final String INVITE_SUBJECT = "Welcome to %s";
-
+  private static final String TASK_SUBJECT = "%s : Task Assignment Notification";
+  private static final String TEST_SUBJECT = "%s : Test Result Notification";
   public static final String INVITE_RANDOM_PWD = "invite-randompwd.ftl";
   public static final String INVITE_CREATE_PWD = "invite-createPassword.ftl";
-
+  public static final String TASK_NOTIFICATION_TEMPLATE = "taskAssignment.ftl";
+  public static final String TEST_NOTIFICATION_TEMPLATE = "testResultStatus.ftl";
   private static EmailUtil INSTANCE = null;
   private SmtpSettings defaultSmtpSettings = null;
   private Mailer mailer = null;
@@ -142,6 +145,45 @@ public class EmailUtil {
     }
   }
 
+  public void sendTaskAssignmentNotificationToUser(
+      String assigneeName, String email, String taskLink, Thread thread, String subject, String templateFilePath)
+      throws IOException, TemplateException {
+    if (defaultSmtpSettings.getEnableSmtpServer()) {
+      Map<String, String> templatePopulator = new HashMap<>();
+      templatePopulator.put("assignee", assigneeName);
+      templatePopulator.put("createdBy", thread.getCreatedBy());
+      templatePopulator.put("taskName", thread.getMessage());
+      templatePopulator.put("taskStatus", thread.getTask().getStatus().toString());
+      templatePopulator.put("taskType", thread.getTask().getType().toString());
+      templatePopulator.put("fieldOldValue", thread.getTask().getOldValue());
+      templatePopulator.put("fieldNewValue", thread.getTask().getSuggestion());
+      templatePopulator.put("taskLink", taskLink);
+
+      sendMail(subject, templatePopulator, email, EMAIL_TEMPLATE_BASEPATH, templateFilePath);
+    }
+  }
+
+  public void sendTestResultEmailNotificationToUser(
+      String email,
+      String testResultLink,
+      String testCaseName,
+      TestCaseResult result,
+      String subject,
+      String templateFilePath)
+      throws IOException, TemplateException {
+    if (defaultSmtpSettings.getEnableSmtpServer()) {
+      Map<String, String> templatePopulator = new HashMap<>();
+      templatePopulator.put("receiverName", email.split("@")[0]);
+      templatePopulator.put("testResultName", testCaseName);
+      templatePopulator.put("testResultDescription", result.getResult());
+      templatePopulator.put("testResultStatus", result.getTestCaseStatus().toString());
+      templatePopulator.put("testResultTimestamp", result.getTimestamp().toString());
+      templatePopulator.put("testResultLink", testResultLink);
+
+      sendMail(subject, templatePopulator, email, EMAIL_TEMPLATE_BASEPATH, templateFilePath);
+    }
+  }
+
   public Email buildEmailWithDefaultSender(EmailRequest request) {
     EmailPopulatingBuilder emailBuilder = EmailBuilder.startingBlank();
     if (request.getRecipientMails() != null
@@ -266,6 +308,14 @@ public class EmailUtil {
     return String.format(INVITE_SUBJECT, defaultSmtpSettings.getEmailingEntity());
   }
 
+  public String getTaskAssignmentSubject() {
+    return String.format(TASK_SUBJECT, defaultSmtpSettings.getEmailingEntity());
+  }
+
+  public String getTestResultSubject() {
+    return String.format(TEST_SUBJECT, defaultSmtpSettings.getEmailingEntity());
+  }
+
   public String getEmailingEntity() {
     return defaultSmtpSettings.getEmailingEntity();
   }

openmetadata-service/src/main/java/org/openmetadata/service/util/NotificationHandler.java

@@ -0,0 +1,273 @@
+package org.openmetadata.service.util;
+
+import static org.openmetadata.schema.settings.SettingsType.TASK_NOTIFICATION_CONFIGURATION;
+import static org.openmetadata.schema.settings.SettingsType.TEST_RESULT_NOTIFICATION_CONFIGURATION;
+import static org.openmetadata.service.Entity.TABLE;
+import static org.openmetadata.service.Entity.TEAM;
+import static org.openmetadata.service.Entity.TEST_CASE;
+import static org.openmetadata.service.Entity.USER;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import freemarker.template.TemplateException;
+import java.io.IOException;
+import java.net.URI;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import javax.ws.rs.container.ContainerResponseContext;
+import javax.ws.rs.core.Response;
+import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.api.configuration.airflow.TaskNotificationConfiguration;
+import org.openmetadata.api.configuration.airflow.TestResultNotificationConfiguration;
+import org.openmetadata.schema.EntityInterface;
+import org.openmetadata.schema.entity.feed.Thread;
+import org.openmetadata.schema.entity.teams.Team;
+import org.openmetadata.schema.entity.teams.User;
+import org.openmetadata.schema.tests.TestCase;
+import org.openmetadata.schema.tests.type.TestCaseResult;
+import org.openmetadata.schema.type.AnnouncementDetails;
+import org.openmetadata.schema.type.ChangeEvent;
+import org.openmetadata.schema.type.EntityReference;
+import org.openmetadata.schema.type.FieldChange;
+import org.openmetadata.schema.type.Post;
+import org.openmetadata.schema.type.Relationship;
+import org.openmetadata.service.Entity;
+import org.openmetadata.service.jdbi3.CollectionDAO;
+import org.openmetadata.service.jdbi3.EntityRepository;
+import org.openmetadata.service.resources.feeds.MessageParser;
+import org.openmetadata.service.resources.settings.SettingsCache;
+import org.openmetadata.service.socket.WebSocketManager;
+
+@Slf4j
+public class NotificationHandler {
+  private final CollectionDAO dao;
+  private final ObjectMapper mapper;
+
+  private final ExecutorService threadScheduler;
+
+  public NotificationHandler(CollectionDAO dao) {
+    this.dao = dao;
+    this.mapper = new ObjectMapper();
+    this.threadScheduler = Executors.newFixedThreadPool(1);
+  }
+
+  public void processNotifications(ContainerResponseContext responseContext) {
+    threadScheduler.submit(
+        () -> {
+          try {
+            handleNotifications(responseContext);
+          } catch (JsonProcessingException e) {
+            LOG.error("[NotificationHandler] Failed to use mapper in converting to Json", e);
+          }
+        });
+  }
+
+  private void handleNotifications(ContainerResponseContext responseContext) throws JsonProcessingException {
+    int responseCode = responseContext.getStatus();
+    if (responseCode == Response.Status.CREATED.getStatusCode()
+        && responseContext.getEntity() != null
+        && responseContext.getEntity().getClass().equals(Thread.class)) {
+      Thread thread = (Thread) responseContext.getEntity();
+      switch (thread.getType()) {
+        case Task:
+          handleTaskNotification(thread);
+          break;
+        case Conversation:
+          handleConversationNotification(thread);
+          break;
+        case Announcement:
+          handleAnnouncementNotification(thread);
+          break;
+      }
+    } else if (responseContext.getEntity() != null
+        && responseContext.getEntity().getClass().equals(ChangeEvent.class)) {
+      ChangeEvent changeEvent = (ChangeEvent) responseContext.getEntity();
+      handleTestResultEmailNotification(changeEvent);
+    }
+  }
+
+  private void handleTaskNotification(Thread thread) throws JsonProcessingException {
+    String jsonThread = mapper.writeValueAsString(thread);
+    if (thread.getPostsCount() == 0) {
+      List<EntityReference> assignees = thread.getTask().getAssignees();
+      HashSet<UUID> receiversList = new HashSet<>();
+      assignees.forEach(
+          e -> {
+            if (Entity.USER.equals(e.getType())) {
+              receiversList.add(e.getId());
+            } else if (Entity.TEAM.equals(e.getType())) {
+              // fetch all that are there in the team
+              List<CollectionDAO.EntityRelationshipRecord> records =
+                  dao.relationshipDAO().findTo(e.getId().toString(), TEAM, Relationship.HAS.ordinal(), Entity.USER);
+              records.forEach((eRecord) -> receiversList.add(eRecord.getId()));
+            }
+          });
+
+      // Send WebSocket Notification
+      WebSocketManager.getInstance()
+          .sendToManyWithUUID(receiversList, WebSocketManager.TASK_BROADCAST_CHANNEL, jsonThread);
+
+      // Send Email Notification If Enabled
+      TaskNotificationConfiguration taskSetting =
+          SettingsCache.getInstance().getSetting(TASK_NOTIFICATION_CONFIGURATION, TaskNotificationConfiguration.class);
+      if (taskSetting.getEnabled()) {
+        handleEmailNotifications(receiversList, thread);
+      }
+    }
+  }
+
+  private void handleAnnouncementNotification(Thread thread) throws JsonProcessingException {
+    String jsonThread = mapper.writeValueAsString(thread);
+    AnnouncementDetails announcementDetails = thread.getAnnouncement();
+    Long currentTimestamp = Instant.now().getEpochSecond();
+    if (announcementDetails.getStartTime() <= currentTimestamp
+        && currentTimestamp <= announcementDetails.getEndTime()) {
+      WebSocketManager.getInstance().broadCastMessageToAll(WebSocketManager.ANNOUNCEMENT_CHANNEL, jsonThread);
+    }
+  }
+
+  private void handleConversationNotification(Thread thread) throws JsonProcessingException {
+    String jsonThread = mapper.writeValueAsString(thread);
+    WebSocketManager.getInstance().broadCastMessageToAll(WebSocketManager.FEED_BROADCAST_CHANNEL, jsonThread);
+    List<MessageParser.EntityLink> mentions;
+    if (thread.getPostsCount() == 0) {
+      mentions = MessageParser.getEntityLinks(thread.getMessage());
+    } else {
+      Post latestPost = thread.getPosts().get(thread.getPostsCount() - 1);
+      mentions = MessageParser.getEntityLinks(latestPost.getMessage());
+    }
+    mentions.forEach(
+        entityLink -> {
+          String fqn = entityLink.getEntityFQN();
+          if (USER.equals(entityLink.getEntityType())) {
+            User user = dao.userDAO().findEntityByName(fqn);
+            WebSocketManager.getInstance().sendToOne(user.getId(), WebSocketManager.MENTION_CHANNEL, jsonThread);
+          } else if (TEAM.equals(entityLink.getEntityType())) {
+            Team team = dao.teamDAO().findEntityByName(fqn);
+            // fetch all that are there in the team
+            List<CollectionDAO.EntityRelationshipRecord> records =
+                dao.relationshipDAO().findTo(team.getId().toString(), TEAM, Relationship.HAS.ordinal(), USER);
+            // Notify on WebSocket for Realtime
+            WebSocketManager.getInstance().sendToManyWithString(records, WebSocketManager.MENTION_CHANNEL, jsonThread);
+          }
+        });
+  }
+
+  private void handleEmailNotifications(HashSet<UUID> userList, Thread thread) {
+    EntityRepository<User> repository = Entity.getEntityRepository(USER);
+    URI urlInstance = thread.getHref();
+    userList.forEach(
+        (id) -> {
+          try {
+            User user = repository.get(null, id, repository.getFields("name,email,href"));
+            EmailUtil.getInstance()
+                .sendTaskAssignmentNotificationToUser(
+                    user.getName(),
+                    user.getEmail(),
+                    String.format(
+                        "%s://%s/users/%s/tasks", urlInstance.getScheme(), urlInstance.getHost(), user.getName()),
+                    thread,
+                    EmailUtil.getInstance().getTaskAssignmentSubject(),
+                    EmailUtil.TASK_NOTIFICATION_TEMPLATE);
+          } catch (IOException ex) {
+            LOG.error("Task Email Notification Failed :", ex);
+          } catch (TemplateException ex) {
+            LOG.error("Task Email Notification Template Parsing Exception :", ex);
+          }
+        });
+  }
+
+  private void handleTestResultEmailNotification(ChangeEvent changeEvent) {
+    if (Objects.nonNull(changeEvent.getChangeDescription())) {
+      FieldChange fieldChange = changeEvent.getChangeDescription().getFieldsUpdated().get(0);
+      String updatedField = fieldChange.getName();
+      if (updatedField.equals("testCaseResult")) {
+        TestCaseResult result = (TestCaseResult) fieldChange.getNewValue();
+        // Send Email Notification If Enabled
+        TestResultNotificationConfiguration testNotificationSetting =
+            SettingsCache.getInstance()
+                .getSetting(TEST_RESULT_NOTIFICATION_CONFIGURATION, TestResultNotificationConfiguration.class);
+        if (testNotificationSetting.getEnabled()
+            && testNotificationSetting.getOnResult().contains(result.getTestCaseStatus())) {
+          List<String> receivers =
+              testNotificationSetting.getReceivers() != null
+                  ? testNotificationSetting.getReceivers()
+                  : new ArrayList<>();
+          if (testNotificationSetting.getSendToOwners()) {
+            EntityInterface entity = (TestCase) changeEvent.getEntity();
+            // Find the Table that have the test case
+            List<CollectionDAO.EntityRelationshipRecord> tableToTestRecord =
+                dao.relationshipDAO()
+                    .findFrom(entity.getId().toString(), TEST_CASE, Relationship.CONTAINS.ordinal(), TABLE);
+            tableToTestRecord.forEach(
+                (tableRecord) -> {
+                  // Find the owners owning the Table , can be a team or Users
+                  List<CollectionDAO.EntityRelationshipRecord> tableOwners =
+                      dao.relationshipDAO()
+                          .findFrom(tableRecord.getId().toString(), TABLE, Relationship.OWNS.ordinal());
+                  tableOwners.forEach(
+                      (owner) -> {
+                        try {
+                          if (USER.equals(owner.getType())) {
+                            User user = dao.userDAO().findEntityById(owner.getId());
+                            receivers.add(user.getEmail());
+                          } else if (TEAM.equals(owner.getType())) {
+                            Team team = dao.teamDAO().findEntityById(owner.getId());
+                            // Fetch the users in the team
+                            List<CollectionDAO.EntityRelationshipRecord> records =
+                                dao.relationshipDAO()
+                                    .findTo(team.getId().toString(), TEAM, Relationship.HAS.ordinal(), USER);
+
+                            records.forEach(
+                                (userRecord) -> {
+                                  try {
+                                    User user = dao.userDAO().findEntityById(userRecord.getId());
+                                    receivers.add(user.getEmail());
+                                  } catch (IOException e) {
+                                    throw new RuntimeException(e);
+                                  }
+                                });
+                          }
+                        } catch (IOException e) {
+                          throw new RuntimeException(e);
+                        }
+                      });
+                });
+          }
+          sendTestResultEmailNotifications(receivers, (TestCase) changeEvent.getEntity(), result);
+        }
+      }
+    }
+  }
+
+  private void sendTestResultEmailNotifications(List<String> emails, TestCase testCase, TestCaseResult result) {
+    emails.forEach(
+        (email) -> {
+          URI urlInstance = testCase.getHref();
+          String testLinkUrl =
+              String.format(
+                  "%s://%s/table/%s/activity_feed",
+                  urlInstance.getScheme(), urlInstance.getHost(), testCase.getEntityFQN());
+          try {
+            EmailUtil.getInstance()
+                .sendTestResultEmailNotificationToUser(
+                    email,
+                    testLinkUrl,
+                    testCase.getName(),
+                    result,
+                    EmailUtil.getInstance().getTestResultSubject(),
+                    EmailUtil.TEST_NOTIFICATION_TEMPLATE);
+          } catch (IOException e) {
+            LOG.error("TestResult Email Notification Failed :", e);
+          } catch (TemplateException e) {
+            LOG.error("Task Email Notification Template Parsing Exception :", e);
+          }
+        });
+  }
+}

openmetadata-service/src/main/resources/emailTemplates/taskAssignment.ftl

@@ -0,0 +1,96 @@
+<!-- [if !mso]>
+<!-->
+<!--![endif]-->
+<!-- Normalize Styles -->
+<!-- [if gte mso 9]>
+<style type="text/css">
+            /* What it does: Normalize space between bullets and text. */
+            /* https://litmus.com/community/discussions/1093-bulletproof-lists-using-ul-and-li */
+            li {
+                text-indent: -1em;
+            }
+</style>
+<![endif]-->
+<div style="display: none; font-size: 1px; line-height: 1px; max-height: 0; max-width: 0; opacity: 0; overflow: hidden; mso-hide: all; font-family: sans-serif;"> &nbsp; </div>
+<table style="background: #F7F8FA; border: 0; border-radius: 0; width: 100%;" cellspacing="0" cellpadding="0">
+  <tbody>
+    <tr>
+      <td class="tw-body" style="padding: 15px 15px 0;" align="center">
+        <table style="background: #F7F8FA; border: 0; border-radius: 0;" cellspacing="0" cellpadding="0">
+          <tbody>
+            <tr>
+              <td class="" style="width: 600px;" align="center">
+                <p style="padding: 5px 5px 5px; font-size: 13px; margin: 0 0 0px; color: #316fea;" align="right"></p>
+                <table style="background: #ffffff; border: 0px; border-radius: 4px; width: 99.6672%; overflow: hidden;" cellspacing="0" cellpadding="0">
+                  <tbody>
+                    <tr>
+                      <td class="" style="padding: 0px; width: 100%;" align="center">
+                        <table style="background: #336f85; border: 0px; border-radius: 0px; width: 599px; height: 53px; margin-left: auto; margin-right: auto;" cellspacing="0" cellpadding="0">
+                          <tbody>
+                            <tr>
+                              <td class="tw-card-header" style="padding: 5px 5px px; width: 366px; color: #ffff; text-decoration: none; font-family: sans-serif;" align="center">
+                                <span style="font-weight: 600;">Task Notification</span>
+                              </td>
+                            </tr>
+                          </tbody>
+                        </table>
+                        <p>
+                          <br />
+                          <br />
+                        </p>
+                        <table dir="ltr" style="border: 0; width: 100%;" cellspacing="0" cellpadding="0">
+                          <tbody>
+                            <tr>
+                              <td class="tw-card-body" style="padding: 20px 35px; text-align: left; color: #6f6f6f; font-family: sans-serif; border-top: 0;">
+                                <h1 class="tw-h1" style="font-size: 24px; font-weight: bold; mso-line-height-rule: exactly; line-height: 32px; margin: 0 0 20px; color: #474747;"> Hello ${assignee},</h1>
+                                <p class="" style="margin: 20px 0; font-size: 16px; mso-line-height-rule: exactly; line-height: 24px;">
+                                  <span style="font-weight: 400;">${createdBy} have assigned you a task to <#if taskType=="UpdateDescription">
+                                      <strong>Update Description </strong>
+                                      <#else>
+                                        <strong>Update Tags </strong>
+                                        </#if>. </span>
+                                  <br />
+                                <h1 class="tw-h1" style="font-size: 20px; font-weight: bold; mso-line-height-rule: exactly; line-height: 10px; margin: 0 0 10px; color: #474747;"> Task Details :</h1>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Task Name :</strong> ${taskName} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Task Status :</strong> ${taskStatus} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Current Value :</strong> ${fieldOldValue} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Suggested Value :</strong> ${fieldNewValue} </span>
+                                <br />
+                                <br />
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <table class="button mobile-w-full" style="border: 0px; border-radius: 7px; margin: 0px auto; width: 525px; background-color: #008bcb; height: 50px;" cellspacing="0" cellpadding="0" align="center">
+                                    <tbody>
+                                      <tr>
+                                        <td class="button__td " style="border-radius: 7px; text-align: center; width: 523px;">
+                                          <!-- [if mso]>
+																					<a href="" class="button__a" target="_blank"
+                                                                                                         style="border-radius: 4px; color: #FFFFFF; display: block; font-family: sans-serif; font-size: 18px; font-weight: bold; mso-height-rule: exactly; line-height: 1.1; padding: 14px 18px; text-decoration: none; text-transform: none; border: 1px solid #316FEA;"></a>
+																					<![endif]-->
+                                          <!-- [if !mso]>
+																					<!-->
+                                          <a class="button__a" style="border-radius: 4px; color: #ffffff; display: block; font-family: sans-serif; font-size: 18px; font-weight: bold; mso-height-rule: exactly; line-height: 1.1; padding: 14px 18px; text-decoration: none; text-transform: none; border: 0;" href="${taskLink}" target="_blank" rel="noopener">View Task</a>
+                                          <!--![endif]-->
+                                        </td>
+                                      </tr>
+                                    </tbody>
+                                  </table>
+                              </td>
+                            </tr>
+                          </tbody>
+                      </td>
+                    </tr>
+                  </tbody>
+                </table>
+              </td>
+            </tr>
+          </tbody>
+        </table>

openmetadata-service/src/main/resources/emailTemplates/testResultStatus.ftl

@@ -0,0 +1,92 @@
+<!-- [if !mso]>
+<!-->
+<!--![endif]-->
+<!-- Normalize Styles -->
+<!-- [if gte mso 9]>
+<style type="text/css">
+            /* What it does: Normalize space between bullets and text. */
+            /* https://litmus.com/community/discussions/1093-bulletproof-lists-using-ul-and-li */
+            li {
+                text-indent: -1em;
+            }
+</style>
+<![endif]-->
+<div style="display: none; font-size: 1px; line-height: 1px; max-height: 0; max-width: 0; opacity: 0; overflow: hidden; mso-hide: all; font-family: sans-serif;"> &nbsp; </div>
+<table style="background: #F7F8FA; border: 0; border-radius: 0; width: 100%;" cellspacing="0" cellpadding="0">
+  <tbody>
+    <tr>
+      <td class="tw-body" style="padding: 15px 15px 0;" align="center">
+        <table style="background: #F7F8FA; border: 0; border-radius: 0;" cellspacing="0" cellpadding="0">
+          <tbody>
+            <tr>
+              <td class="" style="width: 600px;" align="center">
+                <p style="padding: 5px 5px 5px; font-size: 13px; margin: 0 0 0px; color: #316fea;" align="right"></p>
+                <table style="background: #ffffff; border: 0px; border-radius: 4px; width: 99.6672%; overflow: hidden;" cellspacing="0" cellpadding="0">
+                  <tbody>
+                    <tr>
+                      <td class="" style="padding: 0px; width: 100%;" align="center">
+                        <table style="background: #336f85; border: 0px; border-radius: 0px; width: 599px; height: 53px; margin-left: auto; margin-right: auto;" cellspacing="0" cellpadding="0">
+                          <tbody>
+                            <tr>
+                              <td class="tw-card-header" style="padding: 5px 5px px; width: 366px; color: #ffff; text-decoration: none; font-family: sans-serif;" align="center">
+                                <span style="font-weight: 600;">Test Result Notification</span>
+                              </td>
+                            </tr>
+                          </tbody>
+                        </table>
+                        <p>
+                          <br />
+                          <br />
+                        </p>
+                        <table dir="ltr" style="border: 0; width: 100%;" cellspacing="0" cellpadding="0">
+                          <tbody>
+                            <tr>
+                              <td class="tw-card-body" style="padding: 20px 35px; text-align: left; color: #6f6f6f; font-family: sans-serif; border-top: 0;">
+                                <h1 class="tw-h1" style="font-size: 24px; font-weight: bold; mso-line-height-rule: exactly; line-height: 32px; margin: 0 0 20px; color: #474747;"> Hello ${receiverName},</h1>
+                                <p class="" style="margin: 20px 0; font-size: 16px; mso-line-height-rule: exactly; line-height: 24px;">
+                                  <span style="font-weight: 400;">You have a new Test Result Update.</span>
+                                  <br />
+                                <h1 class="tw-h1" style="font-size: 20px; font-weight: bold; mso-line-height-rule: exactly; line-height: 10px; margin: 0 0 10px; color: #474747;"> Task Details :</h1>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Name :</strong> ${testResultName} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Description :</strong> ${testResultDescription} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Status :</strong> ${testResultStatus} </span>
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <strong>Event Timestamp :</strong> ${testResultTimestamp} </span>
+                                <br />
+                                <br />
+                                <br />
+                                <span style="font-weight: 400;">
+                                  <table class="button mobile-w-full" style="border: 0px; border-radius: 7px; margin: 0px auto; width: 525px; background-color: #008bcb; height: 50px;" cellspacing="0" cellpadding="0" align="center">
+                                    <tbody>
+                                      <tr>
+                                        <td class="button__td " style="border-radius: 7px; text-align: center; width: 523px;">
+                                          <!-- [if mso]>
+																						<a href="" class="button__a" target="_blank"
+                                                                                                         style="border-radius: 4px; color: #FFFFFF; display: block; font-family: sans-serif; font-size: 18px; font-weight: bold; mso-height-rule: exactly; line-height: 1.1; padding: 14px 18px; text-decoration: none; text-transform: none; border: 1px solid #316FEA;"></a>
+																						<![endif]-->
+                                          <!-- [if !mso]>
+																						<!-->
+                                          <a class="button__a" style="border-radius: 4px; color: #ffffff; display: block; font-family: sans-serif; font-size: 18px; font-weight: bold; mso-height-rule: exactly; line-height: 1.1; padding: 14px 18px; text-decoration: none; text-transform: none; border: 0;" href="${testResultLink}" target="_blank" rel="noopener">View Activity</a>
+                                          <!--![endif]-->
+                                        </td>
+                                      </tr>
+                                    </tbody>
+                                  </table>
+                              </td>
+                            </tr>
+                          </tbody>
+                      </td>
+                    </tr>
+                  </tbody>
+                </table>
+              </td>
+            </tr>
+          </tbody>
+        </table>

openmetadata-service/src/main/resources/json/data/settings/settingsData.json

@@ -215,5 +215,19 @@
         ]
       }
     ]
+  },
+  {
+    "config_type": "taskNotificationConfiguration",
+    "config_value": {
+      "enabled" : true
+    }
+  },
+  {
+    "config_type": "testResultNotificationConfiguration",
+    "config_value": {
+      "enabled" : false,
+      "onResult": ["Failed", "Aborted"],
+      "sendToOwners": false
+    }
   }
 ]

openmetadata-spec/src/main/resources/json/schema/configuration/taskNotificationConfiguration.json

@@ -0,0 +1,16 @@
+{
+  "$id": "https://open-metadata.org/schema/entity/configuration/taskNotificationConfiguration.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "TaskNotificationConfiguration",
+  "description": "This schema defines the SSL Config.",
+  "type": "object",
+  "javaType": "org.openmetadata.api.configuration.airflow.TaskNotificationConfiguration",
+  "properties": {
+    "enabled": {
+      "description": "Is Task Notification Enabled?",
+      "type" : "boolean",
+      "default": false
+    }
+  },
+  "additionalProperties": false
+}

openmetadata-spec/src/main/resources/json/schema/configuration/testResultNotificationConfiguration.json

@@ -0,0 +1,35 @@
+{
+  "$id": "https://open-metadata.org/schema/entity/configuration/testResultNotificationConfiguration.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "TestResultNotificationConfiguration",
+  "description": "This schema defines the SSL Config.",
+  "type": "object",
+  "javaType": "org.openmetadata.api.configuration.airflow.TestResultNotificationConfiguration",
+  "properties": {
+    "enabled": {
+      "description": "Is Test Notification Enabled?",
+      "type" : "boolean",
+      "default": false
+    },
+    "onResult": {
+      "description": "Send notification on Success, Failed or Aborted?",
+      "type" : "array",
+      "items": {
+        "$ref": "../tests/basic.json#/definitions/testCaseStatus"
+      }
+    },
+    "receivers": {
+      "description": "Send notification on the mail",
+      "type": "array",
+      "items": {
+        "$ref": "../type/basic.json#/definitions/email"
+      }
+    },
+    "sendToOwners": {
+      "description": "Send notification on the mail",
+      "type": "boolean",
+      "default": false
+    }
+  },
+  "additionalProperties": false
+}

openmetadata-spec/src/main/resources/json/schema/settings/settings.json

@@ -22,7 +22,9 @@
         "activityFeedFilterSetting",
         "secretsManagerConfiguration",
         "sandboxModeEnabled",
-        "slackChat"
+        "slackChat",
+        "taskNotificationConfiguration",
+        "testResultNotificationConfiguration"
       ]
     }
   },
@@ -59,7 +61,11 @@
         },
         {
           "$ref": "../configuration/slackEventPubConfiguration.json"
+        },
+        {
+          "$ref": "../configuration/taskNotificationConfiguration.json"
         }
+
       ]
     }
   },

openmetadata-spec/src/main/resources/json/schema/tests/basic.json

@@ -19,15 +19,6 @@
         }
       }
     },
-    "testCaseResult": {
-      "description": "Schema to capture test case result.",
-      "javaType": "org.openmetadata.schema.tests.type.TestCaseResult",
-      "type": "object",
-      "properties": {
-        "timestamp": {
-          "description": "Data one which test case result is taken.",
-          "$ref": "../type/basic.json#/definitions/timestamp"
-        },
     "testCaseStatus": {
       "description": "Status of Test Case run.",
       "javaType": "org.openmetadata.schema.tests.type.TestCaseStatus",
@@ -45,6 +36,19 @@
         }
       ]
     },
+    "testCaseResult": {
+      "description": "Schema to capture test case result.",
+      "javaType": "org.openmetadata.schema.tests.type.TestCaseResult",
+      "type": "object",
+      "properties": {
+        "timestamp": {
+          "description": "Data one which test case result is taken.",
+          "$ref": "../type/basic.json#/definitions/timestamp"
+        },
+        "testCaseStatus": {
+          "description": "Status of Test Case run.",
+          "$ref": "#/definitions/testCaseStatus"
+        },
         "result": {
           "description": "Details of test case results.",
           "type": "string"