Skip to content

Commit

Permalink
Fix owner notification (#16629)
Browse files Browse the repository at this point in the history 
* - Fix Task notification not getting sent to owners

* - Fix Task notification not getting sent to owners

(cherry picked from commit cc2d581)
  • Loading branch information
@mohityadav766
mohityadav766 committed Jun 12, 2024
1 parent 95a51ab commit 521c4fa
Showing 1 changed file with 71 additions and 43 deletions.
114 changes: 71 additions & 43 deletions openmetadata-service/src/main/java/org/openmetadata/service/util/SubscriptionUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -153,42 +153,57 @@ public static Set<String> getOwnerOrFollowers(
}

private static Set<String> getTaskAssignees(
SubscriptionDestination.SubscriptionType type, ChangeEvent event) {
SubscriptionDestination.SubscriptionCategory category,
SubscriptionDestination.SubscriptionType type,
ChangeEvent event) {
Thread thread = AlertsRuleEvaluator.getThread(event);
List<EntityReference> assignees = thread.getTask().getAssignees();
Set<String> receiversList = new HashSet<>();
Map<UUID, Team> teams = new HashMap<>();
Map<UUID, User> users = new HashMap<>();

Team tempTeamVar = null;
User tempUserVar = null;
if (!nullOrEmpty(assignees)) {
for (EntityReference reference : assignees) {
if (Entity.USER.equals(reference.getType())) {
tempUserVar = Entity.getEntity(USER, reference.getId(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(reference.getType())) {
tempTeamVar = Entity.getEntity(TEAM, reference.getId(), "profile", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);

if (category.equals(SubscriptionDestination.SubscriptionCategory.ASSIGNEES)) {
List<EntityReference> assignees = thread.getTask().getAssignees();
if (!nullOrEmpty(assignees)) {
for (EntityReference reference : assignees) {
if (Entity.USER.equals(reference.getType())) {
tempUserVar = Entity.getEntity(USER, reference.getId(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(reference.getType())) {
tempTeamVar = Entity.getEntity(TEAM, reference.getId(), "profile", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
}
}
}
}

for (Post post : thread.getPosts()) {
tempUserVar = Entity.getEntityByName(USER, post.getFrom(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
List<MessageParser.EntityLink> mentions = MessageParser.getEntityLinks(post.getMessage());
for (MessageParser.EntityLink link : mentions) {
if (USER.equals(link.getEntityType())) {
tempUserVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(link.getEntityType())) {
tempTeamVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
for (Post post : thread.getPosts()) {
tempUserVar = Entity.getEntityByName(USER, post.getFrom(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
List<MessageParser.EntityLink> mentions = MessageParser.getEntityLinks(post.getMessage());
for (MessageParser.EntityLink link : mentions) {
if (USER.equals(link.getEntityType())) {
tempUserVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(link.getEntityType())) {
tempTeamVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
}
}
}
}

if (category.equals(SubscriptionDestination.SubscriptionCategory.OWNERS)) {
try {
tempUserVar =
Entity.getEntityByName(USER, thread.getCreatedBy(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} catch (Exception ex) {
LOG.warn("Thread created by unknown user: {}", thread.getCreatedBy());
}
}

// Users
receiversList.addAll(getEmailOrWebhookEndpointForUsers(users.values().stream().toList(), type));

Expand All @@ -199,41 +214,53 @@ private static Set<String> getTaskAssignees(
}

public static Set<String> handleConversationNotification(
SubscriptionDestination.SubscriptionType type, ChangeEvent event) {
SubscriptionDestination.SubscriptionCategory category,
SubscriptionDestination.SubscriptionType type,
ChangeEvent event) {
Thread thread = AlertsRuleEvaluator.getThread(event);
Set<String> receiversList = new HashSet<>();
Map<UUID, Team> teams = new HashMap<>();
Map<UUID, User> users = new HashMap<>();

Team tempTeamVar = null;
User tempUserVar = null;
tempUserVar =
Entity.getEntityByName(USER, thread.getCreatedBy(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
List<MessageParser.EntityLink> mentions = MessageParser.getEntityLinks(thread.getMessage());
for (MessageParser.EntityLink link : mentions) {
if (USER.equals(link.getEntityType())) {
tempUserVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(link.getEntityType())) {
tempTeamVar = Entity.getEntity(link, "", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
}
}

for (Post post : thread.getPosts()) {
tempUserVar = Entity.getEntityByName(USER, post.getFrom(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
mentions = MessageParser.getEntityLinks(post.getMessage());
if (category.equals(SubscriptionDestination.SubscriptionCategory.MENTIONS)) {
List<MessageParser.EntityLink> mentions = MessageParser.getEntityLinks(thread.getMessage());
for (MessageParser.EntityLink link : mentions) {
if (USER.equals(link.getEntityType())) {
tempUserVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(link.getEntityType())) {
tempTeamVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
tempTeamVar = Entity.getEntity(link, "", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
}
}

for (Post post : thread.getPosts()) {
tempUserVar = Entity.getEntityByName(USER, post.getFrom(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
mentions = MessageParser.getEntityLinks(post.getMessage());
for (MessageParser.EntityLink link : mentions) {
if (USER.equals(link.getEntityType())) {
tempUserVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} else if (TEAM.equals(link.getEntityType())) {
tempTeamVar = Entity.getEntity(link, "profile", Include.NON_DELETED);
teams.put(tempTeamVar.getId(), tempTeamVar);
}
}
}
}

if (category.equals(SubscriptionDestination.SubscriptionCategory.OWNERS)) {
try {
tempUserVar =
Entity.getEntityByName(USER, thread.getCreatedBy(), "profile", Include.NON_DELETED);
users.put(tempUserVar.getId(), tempUserVar);
} catch (Exception ex) {
LOG.warn("Thread created by unknown user: {}", thread.getCreatedBy());
}
}

// Users
Expand Down Expand Up @@ -339,8 +366,9 @@ public static Set<String> getTargetsForAlert(
if (event.getEntityType().equals(THREAD)) {
Thread thread = AlertsRuleEvaluator.getThread(event);
switch (thread.getType()) {
case Task -> receiverUrls.addAll(getTaskAssignees(type, event));
case Conversation -> receiverUrls.addAll(handleConversationNotification(type, event));
case Task -> receiverUrls.addAll(getTaskAssignees(category, type, event));
case Conversation -> receiverUrls.addAll(
handleConversationNotification(category, type, event));
// TODO: For Announcement, Immediate Consumer needs to be Notified (find information from
// Lineage)
}
Expand Down

0 comments on commit 521c4fa

Please sign in to comment.