yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-09-27 09:55:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix #471: Topic/Dashboard ownership and follow should reflect in Elastic Search immediately after those actions (#474)

Browse Source 
* Fix #471: Topic/Dashboard ownership and follow should reflect in Elastic Search immediately after those actions
This commit is contained in:
Sriharsha Chintalapani 2021-09-12 21:59:19 -07:00 committed by GitHub
parent 191ecb7332
commit 18b07421ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 104 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
catalog-rest-service/src/main/java/org/openmetadata/catalog/events/ElasticSearchEventHandler.java
View File

@ -26,7 +26,9 @@ import org.elasticsearch.client.RestHighLevelClient;
import org.openmetadata.catalog.CatalogApplicationConfig; import org.openmetadata.catalog.CatalogApplicationConfig;
import org.openmetadata.catalog.ElasticSearchConfiguration; import org.openmetadata.catalog.ElasticSearchConfiguration;
import org.openmetadata.catalog.Entity; import org.openmetadata.catalog.Entity;
import org.openmetadata.catalog.entity.data.Dashboard;
import org.openmetadata.catalog.entity.data.Table; import org.openmetadata.catalog.entity.data.Table;
import org.openmetadata.catalog.entity.data.Topic;
import org.openmetadata.catalog.jdbi3.TableRepository; import org.openmetadata.catalog.jdbi3.TableRepository;
import org.openmetadata.catalog.type.Column; import org.openmetadata.catalog.type.Column;
import org.openmetadata.catalog.type.EntityReference; import org.openmetadata.catalog.type.EntityReference;
@ -74,8 +76,28 @@ public class ElasticSearchEventHandler implements EventHandler {
LOG.info("request Context "+ requestContext.toString()); LOG.info("request Context "+ requestContext.toString());
if (responseContext.getEntity() != null) { if (responseContext.getEntity() != null) {
Object entity = responseContext.getEntity(); Object entity = responseContext.getEntity();
UpdateRequest updateRequest = null;
if (entity.getClass().toString().toLowerCase().endsWith(Entity.TABLE.toLowerCase())) { if (entity.getClass().toString().toLowerCase().endsWith(Entity.TABLE.toLowerCase())) {
Table instance = (Table) entity; Table instance = (Table) entity;
updateRequest = updateTable(instance);
} else if (entity.getClass().toString().toLowerCase().endsWith(Entity.DASHBOARD.toLowerCase())) {
Dashboard instance = (Dashboard) entity;
updateRequest = updateDashboard(instance);
} else if (entity.getClass().toString().toLowerCase().endsWith(Entity.TOPIC.toLowerCase())) {
Topic instance = (Topic) entity;
updateRequest = updateTopic(instance);
}
if (updateRequest != null) {
client.updateAsync(updateRequest, RequestOptions.DEFAULT, listener);
}
}
} catch (Exception e) {
LOG.error("failed to update ES doc", e);
}
return null;
}
private UpdateRequest updateTable(Table instance) {
Map<String, Object> jsonMap = new HashMap<>(); Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("description", instance.getDescription()); jsonMap.put("description", instance.getDescription());
Set<String> tags = new HashSet<>(); Set<String> tags = new HashSet<>();
@ -109,13 +131,62 @@ public class ElasticSearchEventHandler implements EventHandler {
jsonMap.put("last_updated_timestamp", System.currentTimeMillis()); jsonMap.put("last_updated_timestamp", System.currentTimeMillis());
UpdateRequest updateRequest = new UpdateRequest("table_search_index", instance.getId().toString()); UpdateRequest updateRequest = new UpdateRequest("table_search_index", instance.getId().toString());
updateRequest.doc(jsonMap); updateRequest.doc(jsonMap);
client.updateAsync(updateRequest, RequestOptions.DEFAULT, listener); return updateRequest;
} }
private UpdateRequest updateTopic(Topic instance) {
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("description", instance.getDescription());
Set<String> tags = new HashSet<>();
if (instance.getTags() != null) {
instance.getTags().forEach(tag -> tags.add(tag.getTagFQN()));
} }
} catch (Exception e) { if (!tags.isEmpty()) {
LOG.error("failed to update ES doc", e); List<String> tagsList = new ArrayList<>(tags);
jsonMap.put("tags", tagsList);
} }
return null; if(instance.getOwner() != null) {
jsonMap.put("owner", instance.getOwner().getId().toString());
}
if (instance.getFollowers() != null) {
List<String> followers = new ArrayList<>();
for(EntityReference follower: instance.getFollowers()) {
followers.add(follower.getId().toString());
}
jsonMap.put("followers", followers);
}
jsonMap.put("last_updated_timestamp", System.currentTimeMillis());
UpdateRequest updateRequest = new UpdateRequest("topic_search_index", instance.getId().toString());
updateRequest.doc(jsonMap);
return updateRequest;
}
private UpdateRequest updateDashboard(Dashboard instance) {
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("description", instance.getDescription());
Set<String> tags = new HashSet<>();
if (instance.getTags() != null) {
instance.getTags().forEach(tag -> tags.add(tag.getTagFQN()));
}
if (!tags.isEmpty()) {
List<String> tagsList = new ArrayList<>(tags);
jsonMap.put("tags", tagsList);
}
if(instance.getOwner() != null) {
jsonMap.put("owner", instance.getOwner().getId().toString());
}
if (instance.getFollowers() != null) {
List<String> followers = new ArrayList<>();
for(EntityReference follower: instance.getFollowers()) {
followers.add(follower.getId().toString());
}
jsonMap.put("followers", followers);
}
jsonMap.put("last_updated_timestamp", System.currentTimeMillis());
UpdateRequest updateRequest = new UpdateRequest("dashboard_search_index", instance.getId().toString());
updateRequest.doc(jsonMap);
return updateRequest;
} }
public void close() { public void close() {