Add Alert Name to Publishers (#17108)

* Add Alert Name to Publishers

* Fix Test

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java

@@ -122,8 +122,9 @@ public abstract class AbstractEventConsumer
             context.getJobDetail().getJobDataMap().get(DESTINATION_MAP_KEY);
     if (dMap == null) {
       dMap = new HashMap<>();
-      for (SubscriptionDestination subscription : eventSubscription.getDestinations()) {
+      for (SubscriptionDestination subscriptionDest : eventSubscription.getDestinations()) {
-        dMap.put(subscription.getId(), AlertFactory.getAlert(subscription));
+        dMap.put(
+            subscriptionDest.getId(), AlertFactory.getAlert(eventSubscription, subscriptionDest));
       }
       context.getJobDetail().getJobDataMap().put(DESTINATION_MAP_KEY, dMap);
     }

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AlertFactory.java

@@ -2,6 +2,7 @@ package org.openmetadata.service.apps.bundles.changeEvent;
 
 import static org.openmetadata.schema.api.events.CreateEventSubscription.AlertType.ACTIVITY_FEED;
 
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.service.apps.bundles.changeEvent.email.EmailPublisher;
@@ -12,14 +13,15 @@ import org.openmetadata.service.apps.bundles.changeEvent.msteams.MSTeamsPublishe
 import org.openmetadata.service.apps.bundles.changeEvent.slack.SlackEventPublisher;
 
 public class AlertFactory {
-  public static Destination<ChangeEvent> getAlert(SubscriptionDestination config) {
+  public static Destination<ChangeEvent> getAlert(
+      EventSubscription subscription, SubscriptionDestination config) {
     return switch (config.getType()) {
-      case SLACK -> new SlackEventPublisher(config);
+      case SLACK -> new SlackEventPublisher(subscription, config);
-      case MS_TEAMS -> new MSTeamsPublisher(config);
+      case MS_TEAMS -> new MSTeamsPublisher(subscription, config);
-      case G_CHAT -> new GChatPublisher(config);
+      case G_CHAT -> new GChatPublisher(subscription, config);
-      case WEBHOOK -> new GenericPublisher(config);
+      case WEBHOOK -> new GenericPublisher(subscription, config);
-      case EMAIL -> new EmailPublisher(config);
+      case EMAIL -> new EmailPublisher(subscription, config);
-      case ACTIVITY_FEED -> new ActivityFeedPublisher(config);
+      case ACTIVITY_FEED -> new ActivityFeedPublisher(subscription, config);
     };
   }
 }

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/Destination.java

@@ -17,6 +17,7 @@ import static org.openmetadata.schema.entity.events.SubscriptionStatus.Status.AC
 import static org.openmetadata.schema.entity.events.SubscriptionStatus.Status.AWAITING_RETRY;
 import static org.openmetadata.schema.entity.events.SubscriptionStatus.Status.FAILED;
 
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.entity.events.SubscriptionStatus;
 import org.openmetadata.service.events.errors.EventPublisherException;
@@ -27,6 +28,8 @@ public interface Destination<T> {
 
   SubscriptionDestination getSubscriptionDestination();
 
+  EventSubscription getEventSubscriptionForDestination();
+
   void close();
 
   boolean getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/email/EmailPublisher.java

@@ -21,6 +21,7 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.schema.alert.type.EmailAlertConfig;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.service.Entity;
@@ -40,12 +41,15 @@ public class EmailPublisher implements Destination<ChangeEvent> {
   private final CollectionDAO daoCollection;
 
   @Getter private final SubscriptionDestination subscriptionDestination;
+  private final EventSubscription eventSubscription;
 
-  public EmailPublisher(SubscriptionDestination subscription) {
+  public EmailPublisher(
-    if (subscription.getType() == EMAIL) {
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDestination) {
-      this.subscriptionDestination = subscription;
+    if (subscriptionDestination.getType() == EMAIL) {
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDestination;
       this.emailAlertConfig =
-          JsonUtils.convertValue(subscription.getConfig(), EmailAlertConfig.class);
+          JsonUtils.convertValue(subscriptionDestination.getConfig(), EmailAlertConfig.class);
       this.daoCollection = Entity.getCollectionDAO();
     } else {
       throw new IllegalArgumentException("Email Alert Invoked with Illegal Type and Settings.");
@@ -57,9 +61,11 @@ public class EmailPublisher implements Destination<ChangeEvent> {
     try {
       Set<String> receivers =
           getTargetsForAlert(emailAlertConfig, subscriptionDestination.getCategory(), EMAIL, event);
-      EmailMessage emailMessage = emailDecorator.buildOutgoingMessage(event);
+      EmailMessage emailMessage =
+          emailDecorator.buildOutgoingMessage(eventSubscription.getFullyQualifiedName(), event);
       for (String email : receivers) {
-        EmailUtil.sendChangeEventMail(email, emailMessage);
+        EmailUtil.sendChangeEventMail(
+            eventSubscription.getFullyQualifiedName(), email, emailMessage);
       }
       setSuccessStatus(System.currentTimeMillis());
     } catch (Exception e) {
@@ -71,6 +77,11 @@ public class EmailPublisher implements Destination<ChangeEvent> {
     }
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/feed/ActivityFeedPublisher.java

@@ -19,6 +19,7 @@ import static org.openmetadata.schema.entity.events.SubscriptionDestination.Subs
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.entity.feed.Thread;
 import org.openmetadata.schema.type.ChangeEvent;
@@ -38,10 +39,13 @@ public class ActivityFeedPublisher implements Destination<ChangeEvent> {
   FeedRepository feedRepository = new FeedRepository();
 
   @Getter private final SubscriptionDestination subscriptionDestination;
+  private final EventSubscription eventSubscription;
 
-  public ActivityFeedPublisher(SubscriptionDestination subscription) {
+  public ActivityFeedPublisher(
-    if (subscription.getType() == ACTIVITY_FEED) {
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDestination) {
-      this.subscriptionDestination = subscription;
+    if (subscriptionDestination.getType() == ACTIVITY_FEED) {
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDestination;
     } else {
       throw new IllegalArgumentException("Activity Alert Invoked with Illegal Type and Settings.");
     }
@@ -73,6 +77,11 @@ public class ActivityFeedPublisher implements Destination<ChangeEvent> {
     }
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/gchat/GChatPublisher.java

@@ -25,6 +25,7 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.common.utils.CommonUtil;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.Webhook;
@@ -45,13 +46,18 @@ public class GChatPublisher implements Destination<ChangeEvent> {
 
   @Getter private final SubscriptionDestination subscriptionDestination;
 
-  public GChatPublisher(SubscriptionDestination subscription) {
+  private final EventSubscription eventSubscription;
-    if (subscription.getType() == G_CHAT) {
+
-      this.subscriptionDestination = subscription;
+  public GChatPublisher(
-      this.webhook = JsonUtils.convertValue(subscription.getConfig(), Webhook.class);
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDestination) {
+    if (subscriptionDestination.getType() == G_CHAT) {
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDestination;
+      this.webhook = JsonUtils.convertValue(subscriptionDestination.getConfig(), Webhook.class);
 
       // Build Client
-      client = getClient(subscription.getTimeout(), subscription.getReadTimeout());
+      client =
+          getClient(subscriptionDestination.getTimeout(), subscriptionDestination.getReadTimeout());
 
       // Build Target
       if (webhook != null && webhook.getEndpoint() != null) {
@@ -68,7 +74,9 @@ public class GChatPublisher implements Destination<ChangeEvent> {
   @Override
   public void sendMessage(ChangeEvent event) throws EventPublisherException {
     try {
-      GChatMessage gchatMessage = gChatMessageMessageDecorator.buildOutgoingMessage(event);
+      GChatMessage gchatMessage =
+          gChatMessageMessageDecorator.buildOutgoingMessage(
+              eventSubscription.getFullyQualifiedName(), event);
       List<Invocation.Builder> targets =
           getTargetsForWebhookAlert(
               webhook, subscriptionDestination.getCategory(), G_CHAT, client, event);
@@ -86,6 +94,11 @@ public class GChatPublisher implements Destination<ChangeEvent> {
     }
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/generic/GenericPublisher.java

@@ -27,6 +27,7 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.common.utils.CommonUtil;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.Webhook;
@@ -43,14 +44,18 @@ public class GenericPublisher implements Destination<ChangeEvent> {
   private final Webhook webhook;
 
   @Getter private final SubscriptionDestination subscriptionDestination;
+  private final EventSubscription eventSubscription;
 
-  public GenericPublisher(SubscriptionDestination subscription) {
+  public GenericPublisher(
-    if (subscription.getType() == WEBHOOK) {
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDestination) {
-      this.subscriptionDestination = subscription;
+    if (subscriptionDestination.getType() == WEBHOOK) {
-      this.webhook = JsonUtils.convertValue(subscription.getConfig(), Webhook.class);
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDestination;
+      this.webhook = JsonUtils.convertValue(subscriptionDestination.getConfig(), Webhook.class);
 
       // Build Client
-      this.client = getClient(subscription.getTimeout(), subscription.getReadTimeout());
+      this.client =
+          getClient(subscriptionDestination.getTimeout(), subscriptionDestination.getReadTimeout());
     } else {
       throw new IllegalArgumentException(
           "GenericWebhook Alert Invoked with Illegal Type and Settings.");
@@ -104,6 +109,11 @@ public class GenericPublisher implements Destination<ChangeEvent> {
     return SecurityUtil.addHeaders(client.target(webhook.getEndpoint()), authHeaders);
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/msteams/MSTeamsPublisher.java

@@ -26,6 +26,7 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.common.utils.CommonUtil;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.Webhook;
@@ -46,14 +47,18 @@ public class MSTeamsPublisher implements Destination<ChangeEvent> {
   private final Client client;
 
   @Getter private final SubscriptionDestination subscriptionDestination;
+  private final EventSubscription eventSubscription;
 
-  public MSTeamsPublisher(SubscriptionDestination subscription) {
+  public MSTeamsPublisher(
-    if (subscription.getType() == MS_TEAMS) {
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDestination) {
-      this.subscriptionDestination = subscription;
+    if (subscriptionDestination.getType() == MS_TEAMS) {
-      this.webhook = JsonUtils.convertValue(subscription.getConfig(), Webhook.class);
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDestination;
+      this.webhook = JsonUtils.convertValue(subscriptionDestination.getConfig(), Webhook.class);
 
       // Build Client
-      client = getClient(subscription.getTimeout(), subscription.getReadTimeout());
+      client =
+          getClient(subscriptionDestination.getTimeout(), subscriptionDestination.getReadTimeout());
 
       // Build Target
       if (webhook != null && webhook.getEndpoint() != null) {
@@ -70,7 +75,9 @@ public class MSTeamsPublisher implements Destination<ChangeEvent> {
   @Override
   public void sendMessage(ChangeEvent event) throws EventPublisherException {
     try {
-      TeamsMessage teamsMessage = teamsMessageFormatter.buildOutgoingMessage(event);
+      TeamsMessage teamsMessage =
+          teamsMessageFormatter.buildOutgoingMessage(
+              eventSubscription.getFullyQualifiedName(), event);
       List<Invocation.Builder> targets =
           getTargetsForWebhookAlert(
               webhook, subscriptionDestination.getCategory(), MS_TEAMS, client, event);
@@ -97,6 +104,11 @@ public class MSTeamsPublisher implements Destination<ChangeEvent> {
     }
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/slack/SlackEventPublisher.java

@@ -26,6 +26,7 @@ import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.tuple.Pair;
 import org.openmetadata.common.utils.CommonUtil;
+import org.openmetadata.schema.entity.events.EventSubscription;
 import org.openmetadata.schema.entity.events.SubscriptionDestination;
 import org.openmetadata.schema.type.ChangeEvent;
 import org.openmetadata.schema.type.Webhook;
@@ -44,14 +45,17 @@ public class SlackEventPublisher implements Destination<ChangeEvent> {
   private Invocation.Builder target;
   private final Client client;
   @Getter private final SubscriptionDestination subscriptionDestination;
+  private final EventSubscription eventSubscription;
 
-  public SlackEventPublisher(SubscriptionDestination subscription) {
+  public SlackEventPublisher(
-    if (subscription.getType() == SLACK) {
+      EventSubscription eventSubscription, SubscriptionDestination subscriptionDest) {
-      this.subscriptionDestination = subscription;
+    if (subscriptionDest.getType() == SLACK) {
-      this.webhook = JsonUtils.convertValue(subscription.getConfig(), Webhook.class);
+      this.eventSubscription = eventSubscription;
+      this.subscriptionDestination = subscriptionDest;
+      this.webhook = JsonUtils.convertValue(subscriptionDest.getConfig(), Webhook.class);
 
       // Build Client
-      client = getClient(subscription.getTimeout(), subscription.getReadTimeout());
+      client = getClient(subscriptionDest.getTimeout(), subscriptionDest.getReadTimeout());
 
       // Build Target
       if (webhook != null && webhook.getEndpoint() != null) {
@@ -68,7 +72,9 @@ public class SlackEventPublisher implements Destination<ChangeEvent> {
   @Override
   public void sendMessage(ChangeEvent event) throws EventPublisherException {
     try {
-      SlackMessage slackMessage = slackMessageFormatter.buildOutgoingMessage(event);
+      SlackMessage slackMessage =
+          slackMessageFormatter.buildOutgoingMessage(
+              eventSubscription.getFullyQualifiedName(), event);
       List<Invocation.Builder> targets =
           getTargetsForWebhookAlert(
               webhook, subscriptionDestination.getCategory(), SLACK, client, event);
@@ -95,6 +101,11 @@ public class SlackEventPublisher implements Destination<ChangeEvent> {
     }
   }
 
+  @Override
+  public EventSubscription getEventSubscriptionForDestination() {
+    return eventSubscription;
+  }
+
   @Override
   public boolean getEnabled() {
     return subscriptionDestination.getEnabled();

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/EmailMessageDecorator.java

@@ -64,13 +64,13 @@ public class EmailMessageDecorator implements MessageDecorator<EmailMessage> {
   }
 
   @Override
-  public EmailMessage buildEntityMessage(ChangeEvent event) {
+  public EmailMessage buildEntityMessage(String publisherName, ChangeEvent event) {
-    return getEmailMessage(createEntityMessage(event));
+    return getEmailMessage(createEntityMessage(publisherName, event));
   }
 
   @Override
-  public EmailMessage buildThreadMessage(ChangeEvent event) {
+  public EmailMessage buildThreadMessage(String publisherName, ChangeEvent event) {
-    return getEmailMessage(createThreadMessage(event));
+    return getEmailMessage(createThreadMessage(publisherName, event));
   }
 
   public EmailMessage getEmailMessage(OutgoingMessage outgoingMessage) {

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/FeedMessageDecorator.java

@@ -61,12 +61,12 @@ public class FeedMessageDecorator implements MessageDecorator<FeedMessage> {
   }
 
   @Override
-  public FeedMessage buildEntityMessage(ChangeEvent event) {
+  public FeedMessage buildEntityMessage(String publisherName, ChangeEvent event) {
     return null;
   }
 
   @Override
-  public FeedMessage buildThreadMessage(ChangeEvent event) {
+  public FeedMessage buildThreadMessage(String publisherName, ChangeEvent event) {
     return null;
   }
 }

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/GChatMessageDecorator.java

@@ -66,13 +66,13 @@ public class GChatMessageDecorator implements MessageDecorator<GChatMessage> {
   }
 
   @Override
-  public GChatMessage buildEntityMessage(ChangeEvent event) {
+  public GChatMessage buildEntityMessage(String publisherName, ChangeEvent event) {
-    return getGChatMessage(createEntityMessage(event));
+    return getGChatMessage(createEntityMessage(publisherName, event));
   }
 
   @Override
-  public GChatMessage buildThreadMessage(ChangeEvent event) {
+  public GChatMessage buildThreadMessage(String publisherName, ChangeEvent event) {
-    return getGChatMessage(createThreadMessage(event));
+    return getGChatMessage(createThreadMessage(publisherName, event));
   }
 
   private GChatMessage getGChatMessage(OutgoingMessage outgoingMessage) {

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/MSTeamsMessageDecorator.java

@@ -65,13 +65,13 @@ public class MSTeamsMessageDecorator implements MessageDecorator<TeamsMessage> {
   }
 
   @Override
-  public TeamsMessage buildEntityMessage(ChangeEvent event) {
+  public TeamsMessage buildEntityMessage(String publisherName, ChangeEvent event) {
-    return getTeamMessage(createEntityMessage(event));
+    return getTeamMessage(createEntityMessage(publisherName, event));
   }
 
   @Override
-  public TeamsMessage buildThreadMessage(ChangeEvent event) {
+  public TeamsMessage buildThreadMessage(String publisherName, ChangeEvent event) {
-    return getTeamMessage(createThreadMessage(event));
+    return getTeamMessage(createThreadMessage(publisherName, event));
   }
 
   private TeamsMessage getTeamMessage(OutgoingMessage outgoingMessage) {

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/MessageDecorator.java

@@ -64,9 +64,9 @@ public interface MessageDecorator<T> {
 
   String getEntityUrl(String prefix, String fqn, String additionalInput);
 
-  T buildEntityMessage(ChangeEvent event);
+  T buildEntityMessage(String publisherName, ChangeEvent event);
 
-  T buildThreadMessage(ChangeEvent event);
+  T buildThreadMessage(String publisherName, ChangeEvent event);
 
   default String buildEntityUrl(String entityType, EntityInterface entityInterface) {
     String fqn = entityInterface.getFullyQualifiedName();
@@ -140,11 +140,11 @@ public interface MessageDecorator<T> {
     return getEntityUrl(entityType, fqn, activeTab);
   }
 
-  default T buildOutgoingMessage(ChangeEvent event) {
+  default T buildOutgoingMessage(String publisherName, ChangeEvent event) {
     if (event.getEntityType().equals(Entity.THREAD)) {
-      return buildThreadMessage(event);
+      return buildThreadMessage(publisherName, event);
     } else if (Entity.getEntityList().contains(event.getEntityType())) {
-      return buildEntityMessage(event);
+      return buildEntityMessage(publisherName, event);
     } else {
       throw new IllegalArgumentException(
           "Cannot Build Message, Unsupported Entity Type: " + event.getEntityType());
@@ -195,7 +195,7 @@ public interface MessageDecorator<T> {
     return diff;
   }
 
-  default OutgoingMessage createEntityMessage(ChangeEvent event) {
+  default OutgoingMessage createEntityMessage(String publisherName, ChangeEvent event) {
     OutgoingMessage message = new OutgoingMessage();
     message.setUserName(event.getUserName());
     EntityInterface entityInterface = getEntity(event);
@@ -209,13 +209,13 @@ public interface MessageDecorator<T> {
       String headerTxt;
       String headerText;
       if (eventType.equals(Entity.QUERY)) {
-        headerTxt = "%s posted on " + eventType;
+        headerTxt = "[%s] %s posted on " + eventType;
-        headerText = String.format(headerTxt, event.getUserName());
+        headerText = String.format(headerTxt, publisherName, event.getUserName());
       } else {
         String entityUrl = this.buildEntityUrl(event.getEntityType(), entityInterface);
         message.setEntityUrl(entityUrl);
-        headerTxt = "%s posted on " + eventType + " %s";
+        headerTxt = "[%s] %s posted on " + eventType + " %s";
-        headerText = String.format(headerTxt, event.getUserName(), entityUrl);
+        headerText = String.format(headerTxt, publisherName, event.getUserName(), entityUrl);
       }
       message.setHeader(headerText);
     }
@@ -226,7 +226,7 @@ public interface MessageDecorator<T> {
     return message;
   }
 
-  default OutgoingMessage createThreadMessage(ChangeEvent event) {
+  default OutgoingMessage createThreadMessage(String publisherName, ChangeEvent event) {
     OutgoingMessage message = new OutgoingMessage();
     message.setUserName(event.getUserName());
     Thread thread = getThread(event);
@@ -246,12 +246,15 @@ public interface MessageDecorator<T> {
           case THREAD_CREATED -> {
             headerMessage =
                 String.format(
-                    "@%s started a conversation for asset %s", thread.getCreatedBy(), assetUrl);
+                    "[%s] @%s started a conversation for asset %s",
+                    publisherName, thread.getCreatedBy(), assetUrl);
             attachmentList.add(replaceEntityLinks(thread.getMessage()));
           }
           case POST_CREATED -> {
             headerMessage =
-                String.format("@%s posted a message on asset %s", thread.getCreatedBy(), assetUrl);
+                String.format(
+                    "[%s] @%s posted a message on asset %s",
+                    publisherName, thread.getCreatedBy(), assetUrl);
             attachmentList.add(
                 String.format(
                     "@%s : %s", thread.getCreatedBy(), replaceEntityLinks(thread.getMessage())));
@@ -267,8 +270,8 @@ public interface MessageDecorator<T> {
           case THREAD_UPDATED -> {
             headerMessage =
                 String.format(
-                    "@%s posted update on Conversation for asset %s",
+                    "[%s] @%s posted update on Conversation for asset %s",
-                    thread.getUpdatedBy(), assetUrl);
+                    publisherName, thread.getUpdatedBy(), assetUrl);
             attachmentList.add(replaceEntityLinks(thread.getMessage()));
           }
         }
@@ -278,8 +281,8 @@ public interface MessageDecorator<T> {
           case THREAD_CREATED -> {
             headerMessage =
                 String.format(
-                    "@%s created a Task for %s %s",
+                    "[%s] @%s created a Task for %s %s",
-                    thread.getCreatedBy(), entityLink.getEntityType(), assetUrl);
+                    publisherName, thread.getCreatedBy(), entityLink.getEntityType(), assetUrl);
             attachmentList.add(String.format("Task Type : %s", thread.getTask().getType().value()));
             attachmentList.add(
                 String.format(
@@ -293,8 +296,8 @@ public interface MessageDecorator<T> {
           case POST_CREATED -> {
             headerMessage =
                 String.format(
-                    "@%s posted a message on the Task with Id : %s for Asset %s",
+                    "[%s] @%s posted a message on the Task with Id : %s for Asset %s",
-                    thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
+                    publisherName, thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
             thread
                 .getPosts()
                 .forEach(
@@ -307,8 +310,8 @@ public interface MessageDecorator<T> {
           case THREAD_UPDATED -> {
             headerMessage =
                 String.format(
-                    "@%s posted update on the Task with Id : %s for Asset %s",
+                    "[%s] @%s posted update on the Task with Id : %s for Asset %s",
-                    thread.getUpdatedBy(), thread.getTask().getId(), assetUrl);
+                    publisherName, thread.getUpdatedBy(), thread.getTask().getId(), assetUrl);
             attachmentList.add(String.format("Task Type : %s", thread.getTask().getType().value()));
             attachmentList.add(
                 String.format(
@@ -322,15 +325,15 @@ public interface MessageDecorator<T> {
           case TASK_CLOSED -> {
             headerMessage =
                 String.format(
-                    "@%s closed Task with Id : %s for Asset %s",
+                    "[%s] @%s closed Task with Id : %s for Asset %s",
-                    thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
+                    publisherName, thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
             attachmentList.add(String.format("Current Status : %s", thread.getTask().getStatus()));
           }
           case TASK_RESOLVED -> {
             headerMessage =
                 String.format(
-                    "@%s resolved Task with Id : %s for Asset %s",
+                    "[%s] @%s resolved Task with Id : %s for Asset %s",
-                    thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
+                    publisherName, thread.getCreatedBy(), thread.getTask().getId(), assetUrl);
             attachmentList.add(String.format("Current Status : %s", thread.getTask().getStatus()));
           }
         }
@@ -339,7 +342,9 @@ public interface MessageDecorator<T> {
         switch (event.getEventType()) {
           case THREAD_CREATED -> {
             headerMessage =
-                String.format("**@%s** posted an **Announcement**", thread.getCreatedBy());
+                String.format(
+                    "[%s] **@%s** posted an **Announcement**",
+                    publisherName, thread.getCreatedBy());
             attachmentList.add(
                 String.format("Description : %s", thread.getAnnouncement().getDescription()));
             attachmentList.add(
@@ -365,7 +370,8 @@ public interface MessageDecorator<T> {
           case THREAD_UPDATED -> {
             headerMessage =
                 String.format(
-                    "**@%s** posted an update on  **Announcement**", thread.getUpdatedBy());
+                    "[%s] **@%s** posted an update on  **Announcement**",
+                    publisherName, thread.getUpdatedBy());
             attachmentList.add(
                 String.format("Description : %s", thread.getAnnouncement().getDescription()));
             attachmentList.add(

openmetadata-service/src/main/java/org/openmetadata/service/formatter/decorators/SlackMessageDecorator.java

@@ -66,13 +66,13 @@ public class SlackMessageDecorator implements MessageDecorator<SlackMessage> {
   }
 
   @Override
-  public SlackMessage buildEntityMessage(ChangeEvent event) {
+  public SlackMessage buildEntityMessage(String publisherName, ChangeEvent event) {
-    return getSlackMessage(createEntityMessage(event));
+    return getSlackMessage(createEntityMessage(publisherName, event));
   }
 
   @Override
-  public SlackMessage buildThreadMessage(ChangeEvent event) {
+  public SlackMessage buildThreadMessage(String publisherName, ChangeEvent event) {
-    return getSlackMessage(createThreadMessage(event));
+    return getSlackMessage(createThreadMessage(publisherName, event));
   }
 
   private SlackMessage getSlackMessage(OutgoingMessage outgoingMessage) {

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

@@ -71,7 +71,7 @@ public class EmailUtil {
   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 CHANGE_EVENT_UPDATE = "Change Event Update from %s";
+  private static final String CHANGE_EVENT_UPDATE = "[%s] - Change Event Update from %s";
 
   private static final String TASK_SUBJECT = "%s : Task Assignment Notification";
   public static final String INVITE_RANDOM_PWD = "invite-randompwd.ftl";
@@ -306,7 +306,8 @@ public class EmailUtil {
     }
   }
 
-  public static void sendChangeEventMail(String receiverMail, EmailMessage emailMessaged) {
+  public static void sendChangeEventMail(
+      String publisherName, String receiverMail, EmailMessage emailMessaged) {
     if (Boolean.TRUE.equals(getSmtpSettings().getEnableSmtpServer())) {
       Map<String, Object> templatePopulator = new HashMap<>();
       templatePopulator.put(EmailUtil.USERNAME, receiverMail.split("@")[0]);
@@ -320,7 +321,7 @@ public class EmailUtil {
       templatePopulator.put("changeMessage", buff.toString());
       try {
         EmailUtil.sendMail(
-            EmailUtil.getChangeEventTemplate(),
+            EmailUtil.getChangeEventTemplate(publisherName),
             templatePopulator,
             receiverMail,
             EmailUtil.EMAIL_TEMPLATE_BASEPATH,
@@ -407,8 +408,8 @@ public class EmailUtil {
     return String.format(INVITE_SUBJECT, getSmtpSettings().getEmailingEntity());
   }
 
-  public static String getChangeEventTemplate() {
+  public static String getChangeEventTemplate(String publisherName) {
-    return String.format(CHANGE_EVENT_UPDATE, getSmtpSettings().getEmailingEntity());
+    return String.format(CHANGE_EVENT_UPDATE, publisherName, getSmtpSettings().getEmailingEntity());
   }
 
   public static String getTaskAssignmentSubject() {

openmetadata-service/src/test/java/org/openmetadata/service/resources/events/EventSubscriptionResourceTest.java

@@ -1644,7 +1644,11 @@ public class EventSubscriptionResourceTest
   private String buildExpectedTextFormatSlack(EventSubscription alert) {
     String updatedBy = alert.getUpdatedBy();
     return String.format(
-        "%s posted on " + Entity.EVENT_SUBSCRIPTION + " %s", updatedBy, getEntityUrlSlack(alert));
+        "[%s] %s posted on %s %s",
+        alert.getFullyQualifiedName(),
+        updatedBy,
+        Entity.EVENT_SUBSCRIPTION,
+        getEntityUrlSlack(alert));
   }
 
   private String getEntityUrlSlack(EventSubscription alert) {
@@ -1692,8 +1696,12 @@ public class EventSubscriptionResourceTest
   private String buildExpectedActivityTitleTextFormatMSTeams(EventSubscription alert) {
     String updatedBy = alert.getUpdatedBy();
     return String.format(
-        "%s posted on %s [\"%s\"](/%s)",
+        "[%s] %s posted on %s [\"%s\"](/%s)",
-        updatedBy, Entity.EVENT_SUBSCRIPTION, alert.getName(), getEntityUrlMSTeams());
+        alert.getFullyQualifiedName(),
+        updatedBy,
+        Entity.EVENT_SUBSCRIPTION,
+        alert.getName(),
+        getEntityUrlMSTeams());
   }
 
   private String getEntityUrlMSTeams() {