You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by ra...@apache.org on 2026/09/08 05:44:42 UTC
(ranger) 01/01: Add Metadata Registry Kafka source for Ranger Tag Sync.
This is an automated email from the ASF dual-hosted git repository.
ramackri pushed a commit to branch feature/metadataregistry-tag-sync
in repository https://gitbox.apache.org/repos/asf/ranger.git
commit ec1e3e2b88f515ac4bee6e7ae781143512171ff7
Author: ramk <ra...@cloudera.com>
AuthorDate: Tue Sep 8 11:14:28 2026 +0530
Add Metadata Registry Kafka source for Ranger Tag Sync.
Consume udf.governance entity notifications and map URN-based Hive resources to Ranger service tags, mirroring the existing Atlas Tag Sync flow.
Co-authored-by: Cursor <cu...@cursor.com>
---
.../ranger/tagsync/process/TagSynchronizer.java | 2 +
.../metadataregistry/MetadataRegistryEntity.java | 75 +++++++
.../MetadataRegistryEntityWithTags.java | 28 +++
.../MetadataRegistryHiveResourceMapper.java | 106 +++++++++
.../MetadataRegistryNotificationFields.java | 26 +++
.../MetadataRegistryNotificationMapper.java | 124 +++++++++++
.../MetadataRegistryNotificationParser.java | 160 ++++++++++++++
.../MetadataRegistryNotificationWrapper.java | 52 +++++
.../MetadataRegistryResourceMapper.java | 68 ++++++
.../MetadataRegistryResourceMapperUtil.java | 93 ++++++++
.../MetadataRegistryTagSource.java | 243 +++++++++++++++++++++
tagsync/src/main/resources/ranger-tagsync-site.xml | 20 ++
.../TestMetadataRegistryHiveResourceMapper.java | 73 +++++++
13 files changed, 1070 insertions(+)
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
index 3bb5e2e6d..5110c5883 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
@@ -405,6 +405,8 @@ private static TagSource getTagSourceFromConfig(Properties props, String propPre
className = "org.apache.ranger.tagsync.source.atlas.AtlasTagSource";
} else if (tagSourceName.equals("atlasrest")) {
className = "org.apache.ranger.tagsync.source.atlasrest.AtlasRESTTagSource";
+ } else if (tagSourceName.equalsIgnoreCase("metadataregistry")) {
+ className = "org.apache.ranger.tagsync.source.metadataregistry.MetadataRegistryTagSource";
} else {
LOG.error("tagSource name doesn't have any class associated with it. tagSourceName={}, propertyPrefix={}", tagSourceName, propPrefix);
}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntity.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntity.java
new file mode 100644
index 000000000..33f9e92ce
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntity.java
@@ -0,0 +1,75 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+/**
+ * Entity identity from Metadata Registry {@code udf.governance} notifications.
+ */
+public class MetadataRegistryEntity {
+ private final String urn;
+ private final Long entityId;
+ private final String entityType;
+ private final String fieldPath;
+ private final String name;
+ private final String displayString;
+ private final String deployment;
+ private final String technology;
+
+ public MetadataRegistryEntity(
+ String urn,
+ Long entityId,
+ String entityType,
+ String fieldPath,
+ String name,
+ String displayString,
+ String deployment,
+ String technology) {
+ this.urn = urn;
+ this.entityId = entityId;
+ this.entityType = entityType;
+ this.fieldPath = fieldPath;
+ this.name = name;
+ this.displayString = displayString;
+ this.deployment = deployment;
+ this.technology = technology;
+ }
+
+ public String getUrn() {
+ return urn;
+ }
+
+ public Long getEntityId() {
+ return entityId;
+ }
+
+ public String getEntityType() {
+ return entityType;
+ }
+
+ public String getFieldPath() {
+ return fieldPath;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getDisplayString() {
+ return displayString;
+ }
+
+ public String getDeployment() {
+ return deployment;
+ }
+
+ public String getTechnology() {
+ return technology;
+ }
+
+ public String getResourceKey() {
+ return urn != null ? urn : String.valueOf(entityId);
+ }
+
+ @Override
+ public String toString() {
+ return "{urn=" + urn + ", entityType=" + entityType + ", fieldPath=" + fieldPath + "}";
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntityWithTags.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntityWithTags.java
new file mode 100644
index 000000000..2688a5e53
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryEntityWithTags.java
@@ -0,0 +1,28 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.List;
+
+public class MetadataRegistryEntityWithTags {
+ private final MetadataRegistryEntity entity;
+ private final List<MetadataRegistryNotificationParser.MetadataRegistryClassification> tags;
+
+ public MetadataRegistryEntityWithTags(MetadataRegistryNotificationWrapper notification) {
+ this.entity = notification.getEntity();
+ this.tags = notification.getClassifications();
+ }
+
+ public MetadataRegistryEntity getEntity() {
+ return entity;
+ }
+
+ public List<MetadataRegistryNotificationParser.MetadataRegistryClassification> getTags() {
+ return tags;
+ }
+
+ @Override
+ public String toString() {
+ return "{entity=" + entity + ", tags=" + (CollectionUtils.isEmpty(tags) ? "[]" : tags) + "}";
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryHiveResourceMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryHiveResourceMapper.java
new file mode 100644
index 000000000..ea0d2bea6
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryHiveResourceMapper.java
@@ -0,0 +1,106 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.ranger.plugin.model.RangerPolicy.RangerPolicyResource;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Maps Hive entities from Metadata Registry URNs to Ranger Hive service resources.
+ *
+ * <p>Example URN: {@code urn:hive:table::hms:prod:sales:orders}
+ */
+public class MetadataRegistryHiveResourceMapper extends MetadataRegistryResourceMapper {
+ public static final String ENTITY_TYPE_HIVE_DATABASE = "hive:database";
+ public static final String ENTITY_TYPE_HIVE_TABLE = "hive:table";
+
+ public static final String RANGER_TYPE_HIVE_DB = "database";
+ public static final String RANGER_TYPE_HIVE_TABLE = "table";
+ public static final String RANGER_TYPE_HIVE_COLUMN = "column";
+
+ public static final String FIELD_PATH_COLUMNS_PREFIX = "columns.";
+
+ private static final String URN_CANONICAL_DELIMITER = "::";
+
+ public MetadataRegistryHiveResourceMapper() {
+ super("hive", new String[] {ENTITY_TYPE_HIVE_DATABASE, ENTITY_TYPE_HIVE_TABLE});
+ }
+
+ @Override
+ public RangerServiceResource buildResource(MetadataRegistryEntity entity) throws Exception {
+ if (entity == null || StringUtils.isBlank(entity.getUrn())) {
+ throw new Exception("Metadata Registry entity URN is required");
+ }
+
+ String[] segments = canonicalSegments(entity.getUrn());
+ if (segments.length < 3) {
+ throw new Exception("Hive entity URN missing deployment segments: " + entity.getUrn());
+ }
+
+ String deployment = entity.getDeployment() != null ? entity.getDeployment() : segments[1];
+ String serviceName = getRangerServiceName(deployment);
+ String resourceKey = entity.getResourceKey();
+
+ Map<String, RangerPolicyResource> elements = new HashMap<>();
+
+ if (isColumnFieldPath(entity.getFieldPath())) {
+ if (!ENTITY_TYPE_HIVE_TABLE.equals(entity.getEntityType())) {
+ throw new Exception("Column fieldPath requires hive:table entity: " + entity.getUrn());
+ }
+ if (segments.length < 4) {
+ throw new Exception("Hive table URN missing database/table segments: " + entity.getUrn());
+ }
+ String database = segments[2];
+ String table = entity.getName() != null ? entity.getName() : segments[3];
+ String column = columnName(entity.getFieldPath());
+ elements.put(RANGER_TYPE_HIVE_DB, new RangerPolicyResource(database));
+ elements.put(RANGER_TYPE_HIVE_TABLE, new RangerPolicyResource(table));
+ elements.put(RANGER_TYPE_HIVE_COLUMN, new RangerPolicyResource(column));
+ } else if (ENTITY_TYPE_HIVE_DATABASE.equals(entity.getEntityType())) {
+ String database = entity.getName() != null ? entity.getName() : segments[segments.length - 1];
+ elements.put(RANGER_TYPE_HIVE_DB, new RangerPolicyResource(database));
+ } else if (ENTITY_TYPE_HIVE_TABLE.equals(entity.getEntityType())) {
+ if (segments.length < 4) {
+ throw new Exception("Hive table URN missing database/table segments: " + entity.getUrn());
+ }
+ String database = segments[2];
+ String table = entity.getName() != null ? entity.getName() : segments[3];
+ elements.put(RANGER_TYPE_HIVE_DB, new RangerPolicyResource(database));
+ elements.put(RANGER_TYPE_HIVE_TABLE, new RangerPolicyResource(table));
+ } else {
+ throw new Exception("Unsupported Metadata Registry entity type for Hive mapper: " + entity.getEntityType());
+ }
+
+ if (elements.isEmpty()) {
+ throw new Exception("Unable to derive Hive resource elements from " + entity.getUrn());
+ }
+
+ return new RangerServiceResource(resourceKey, serviceName, elements);
+ }
+
+ static String[] canonicalSegments(String urn) throws Exception {
+ int delimiter = urn.indexOf(URN_CANONICAL_DELIMITER);
+ if (delimiter < 0 || delimiter + URN_CANONICAL_DELIMITER.length() >= urn.length()) {
+ throw new Exception("Malformed Metadata Registry URN: " + urn);
+ }
+ String canonicalPath = urn.substring(delimiter + URN_CANONICAL_DELIMITER.length());
+ return canonicalPath.split(":", -1);
+ }
+
+ static boolean isColumnFieldPath(String fieldPath) {
+ return fieldPath != null
+ && fieldPath.toLowerCase(Locale.ROOT).startsWith(FIELD_PATH_COLUMNS_PREFIX);
+ }
+
+ static String columnName(String fieldPath) {
+ if (!isColumnFieldPath(fieldPath)) {
+ return null;
+ }
+ String remainder = fieldPath.substring(FIELD_PATH_COLUMNS_PREFIX.length()).trim();
+ int dot = remainder.indexOf('.');
+ return dot >= 0 ? remainder.substring(0, dot) : remainder;
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationFields.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationFields.java
new file mode 100644
index 000000000..712cf9ce2
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationFields.java
@@ -0,0 +1,26 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+/**
+ * JSON field names in Metadata Registry governance notification payloads.
+ */
+public final class MetadataRegistryNotificationFields {
+ public static final String MESSAGE = "message";
+ public static final String MESSAGE_TYPE = "type";
+ public static final String MESSAGE_TYPE_VALUE = "METADATA_REGISTRY_ENTITY_NOTIFICATION";
+ public static final String OPERATION_TYPE = "operationType";
+ public static final String ENTITY = "entity";
+
+ public static final String URN = "urn";
+ public static final String ENTITY_ID = "entityId";
+ public static final String ENTITY_TYPE = "entityType";
+ public static final String FIELD_PATH = "fieldPath";
+ public static final String NAME = "name";
+ public static final String DISPLAY_STRING = "displayString";
+ public static final String DEPLOYMENT = "deployment";
+ public static final String TECHNOLOGY = "technology";
+ public static final String CLASSIFICATIONS = "classifications";
+ public static final String TYPE_NAME = "typeName";
+ public static final String ATTRIBUTES = "attributes";
+
+ private MetadataRegistryNotificationFields() {}
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationMapper.java
new file mode 100644
index 000000000..6142d7d7b
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationMapper.java
@@ -0,0 +1,124 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+import org.apache.ranger.plugin.model.RangerTag;
+import org.apache.ranger.plugin.model.RangerTagDef;
+import org.apache.ranger.plugin.model.RangerTagDef.RangerTagAttributeDef;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Maps Metadata Registry notifications to Ranger {@link ServiceTags}, following Atlas Tag Sync flow.
+ */
+public final class MetadataRegistryNotificationMapper {
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataRegistryNotificationMapper.class);
+
+ private MetadataRegistryNotificationMapper() {}
+
+ public static boolean isNotificationHandled(MetadataRegistryNotificationWrapper notification) {
+ if (notification == null || notification.getOpType() == null) {
+ return false;
+ }
+
+ boolean ret = switch (notification.getOpType()) {
+ case ENTITY_CREATE, ENTITY_UPDATE ->
+ !notification.isEmptyClassifications();
+ case ENTITY_DELETE -> true;
+ case CLASSIFICATION_ADD, CLASSIFICATION_UPDATE, CLASSIFICATION_DELETE -> true;
+ default -> false;
+ };
+
+ if (ret) {
+ ret = notification.isEntityTypeHandled();
+ }
+
+ return ret;
+ }
+
+ public static void logUnhandledNotification(MetadataRegistryNotificationWrapper notification) {
+ if (notification == null) {
+ return;
+ }
+ if (!notification.isEntityTypeHandled()) {
+ LOG.warn("Tag Sync is not enabled to handle Metadata Registry entity type [{}]",
+ notification.getEntity().getEntityType());
+ }
+ }
+
+ public static Map<String, ServiceTags> processEntities(List<MetadataRegistryEntityWithTags> entitiesWithTags) {
+ Map<String, ServiceTags> ret = new HashMap<>();
+ if (CollectionUtils.isEmpty(entitiesWithTags)) {
+ return ret;
+ }
+ for (MetadataRegistryEntityWithTags element : entitiesWithTags) {
+ if (element.getEntity() != null) {
+ buildServiceTags(element, ret);
+ }
+ }
+ if (MapUtils.isNotEmpty(ret)) {
+ for (ServiceTags serviceTags : ret.values()) {
+ serviceTags.setOp(ServiceTags.OP_REPLACE);
+ }
+ }
+ return ret;
+ }
+
+ private static void buildServiceTags(
+ MetadataRegistryEntityWithTags entityWithTags,
+ Map<String, ServiceTags> serviceTagsMap) {
+ MetadataRegistryEntity entity = entityWithTags.getEntity();
+ RangerServiceResource serviceResource = MetadataRegistryResourceMapperUtil.getRangerServiceResource(entity);
+ if (serviceResource == null) {
+ LOG.error("Failed to build serviceResource for Metadata Registry entity {}", entity.getUrn());
+ return;
+ }
+
+ String serviceName = serviceResource.getServiceName();
+ ServiceTags serviceTags = serviceTagsMap.computeIfAbsent(serviceName, ignored -> {
+ ServiceTags created = new ServiceTags();
+ created.setOp(ServiceTags.OP_ADD_OR_UPDATE);
+ created.setServiceName(serviceName);
+ return created;
+ });
+
+ serviceResource.setId((long) serviceTags.getServiceResources().size());
+ serviceTags.getServiceResources().add(serviceResource);
+
+ List<Long> tagIds = new ArrayList<>();
+ List<MetadataRegistryNotificationParser.MetadataRegistryClassification> tags = entityWithTags.getTags();
+ if (CollectionUtils.isNotEmpty(tags)) {
+ for (MetadataRegistryNotificationParser.MetadataRegistryClassification tag : tags) {
+ RangerTag rangerTag = new RangerTag(
+ null,
+ tag.getTagName(),
+ new HashMap<>(tag.getAttributes()),
+ RangerTag.OWNER_SERVICERESOURCE);
+ rangerTag.setId((long) serviceTags.getTags().size());
+ serviceTags.getTags().put(rangerTag.getId(), rangerTag);
+ tagIds.add(rangerTag.getId());
+
+ RangerTagDef tagDef = new RangerTagDef(tag.getTagName(), "MetadataRegistry");
+ if (MapUtils.isNotEmpty(tag.getAttributes())) {
+ List<RangerTagAttributeDef> attributeDefs = new ArrayList<>();
+ for (String attributeName : tag.getAttributes().keySet()) {
+ attributeDefs.add(new RangerTagAttributeDef(attributeName, "string"));
+ }
+ tagDef.setAttributeDefs(attributeDefs);
+ }
+ tagDef.setId((long) serviceTags.getTagDefinitions().size());
+ serviceTags.getTagDefinitions().put(tagDef.getId(), tagDef);
+ }
+ }
+
+ serviceTags.getResourceToTagIds().put(serviceResource.getId(), tagIds);
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationParser.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationParser.java
new file mode 100644
index 000000000..b0752a2d1
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationParser.java
@@ -0,0 +1,160 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.ranger.authorization.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Parses Metadata Registry {@code udf.governance} Kafka payloads.
+ */
+public final class MetadataRegistryNotificationParser {
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataRegistryNotificationParser.class);
+
+ private MetadataRegistryNotificationParser() {}
+
+ public static MetadataRegistryNotificationWrapper parse(String payload) {
+ if (StringUtils.isBlank(payload)) {
+ return null;
+ }
+ try {
+ JsonNode root = JsonUtils.getMapper().readTree(payload);
+ JsonNode message = root.path(MetadataRegistryNotificationFields.MESSAGE);
+ if (message.isMissingNode()) {
+ LOG.warn("Metadata Registry notification missing message node");
+ return null;
+ }
+ if (!MetadataRegistryNotificationFields.MESSAGE_TYPE_VALUE.equals(
+ message.path(MetadataRegistryNotificationFields.MESSAGE_TYPE).asText(null))) {
+ LOG.debug("Ignoring non Metadata Registry notification type");
+ return null;
+ }
+
+ String operationType = message.path(MetadataRegistryNotificationFields.OPERATION_TYPE).asText(null);
+ NotificationOpType opType = toOpType(operationType);
+ if (opType == NotificationOpType.UNKNOWN) {
+ LOG.warn("Unsupported Metadata Registry operationType={}", operationType);
+ return null;
+ }
+
+ JsonNode entityNode = message.path(MetadataRegistryNotificationFields.ENTITY);
+ MetadataRegistryEntity entity = toEntity(entityNode);
+ if (entity == null) {
+ LOG.warn("Metadata Registry notification missing entity");
+ return null;
+ }
+
+ List<MetadataRegistryClassification> classifications = toClassifications(
+ entityNode.path(MetadataRegistryNotificationFields.CLASSIFICATIONS));
+
+ return new MetadataRegistryNotificationWrapper(entity, classifications, opType);
+ } catch (Exception exception) {
+ LOG.error("Failed to parse Metadata Registry notification payload", exception);
+ return null;
+ }
+ }
+
+ private static NotificationOpType toOpType(String operationType) {
+ if (operationType == null) {
+ return NotificationOpType.UNKNOWN;
+ }
+ try {
+ return NotificationOpType.valueOf(operationType);
+ } catch (IllegalArgumentException exception) {
+ return NotificationOpType.UNKNOWN;
+ }
+ }
+
+ private static MetadataRegistryEntity toEntity(JsonNode entityNode) {
+ if (entityNode == null || entityNode.isMissingNode()) {
+ return null;
+ }
+ String urn = entityNode.path(MetadataRegistryNotificationFields.URN).asText(null);
+ if (StringUtils.isBlank(urn)) {
+ return null;
+ }
+ Long entityId = entityNode.hasNonNull(MetadataRegistryNotificationFields.ENTITY_ID)
+ ? entityNode.path(MetadataRegistryNotificationFields.ENTITY_ID).asLong()
+ : null;
+ return new MetadataRegistryEntity(
+ urn,
+ entityId,
+ entityNode.path(MetadataRegistryNotificationFields.ENTITY_TYPE).asText(null),
+ entityNode.path(MetadataRegistryNotificationFields.FIELD_PATH).asText(""),
+ entityNode.path(MetadataRegistryNotificationFields.NAME).asText(null),
+ entityNode.path(MetadataRegistryNotificationFields.DISPLAY_STRING).asText(null),
+ entityNode.path(MetadataRegistryNotificationFields.DEPLOYMENT).asText(null),
+ entityNode.path(MetadataRegistryNotificationFields.TECHNOLOGY).asText(null));
+ }
+
+ private static List<MetadataRegistryClassification> toClassifications(JsonNode classificationsNode) {
+ List<MetadataRegistryClassification> classifications = new ArrayList<>();
+ if (classificationsNode == null || !classificationsNode.isArray()) {
+ return classifications;
+ }
+ for (JsonNode node : classificationsNode) {
+ String typeName = node.path(MetadataRegistryNotificationFields.TYPE_NAME).asText(null);
+ if (StringUtils.isBlank(typeName)) {
+ continue;
+ }
+ Map<String, String> attributes = new HashMap<>();
+ JsonNode attrsNode = node.path(MetadataRegistryNotificationFields.ATTRIBUTES);
+ if (attrsNode.isObject()) {
+ Iterator<Map.Entry<String, JsonNode>> fields = attrsNode.fields();
+ while (fields.hasNext()) {
+ Map.Entry<String, JsonNode> entry = fields.next();
+ if (!entry.getValue().isNull()) {
+ attributes.put(entry.getKey(), entry.getValue().asText());
+ }
+ }
+ }
+ classifications.add(new MetadataRegistryClassification(typeName, attributes));
+ }
+ return classifications;
+ }
+
+ public enum NotificationOpType {
+ ENTITY_CREATE,
+ ENTITY_UPDATE,
+ ENTITY_DELETE,
+ CLASSIFICATION_ADD,
+ CLASSIFICATION_UPDATE,
+ CLASSIFICATION_DELETE,
+ UNKNOWN
+ }
+
+ public static final class MetadataRegistryClassification {
+ private final String typeName;
+ private final Map<String, String> attributes;
+
+ public MetadataRegistryClassification(String typeName, Map<String, String> attributes) {
+ this.typeName = typeName;
+ this.attributes = attributes == null ? Map.of() : Map.copyOf(attributes);
+ }
+
+ public String getTypeName() {
+ return typeName;
+ }
+
+ public Map<String, String> getAttributes() {
+ return attributes;
+ }
+
+ public String getTagName() {
+ if (typeName == null) {
+ return "UNKNOWN";
+ }
+ int colon = typeName.indexOf(':');
+ return colon >= 0 ? typeName.substring(colon + 1).toUpperCase() : typeName.toUpperCase();
+ }
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationWrapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationWrapper.java
new file mode 100644
index 000000000..ed84be74c
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryNotificationWrapper.java
@@ -0,0 +1,52 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.List;
+
+public class MetadataRegistryNotificationWrapper {
+ private final MetadataRegistryEntity entity;
+ private final List<MetadataRegistryNotificationParser.MetadataRegistryClassification> classifications;
+ private final MetadataRegistryNotificationParser.NotificationOpType opType;
+ private final boolean entityTypeHandled;
+ private final boolean emptyClassifications;
+
+ public MetadataRegistryNotificationWrapper(
+ MetadataRegistryEntity entity,
+ List<MetadataRegistryNotificationParser.MetadataRegistryClassification> classifications,
+ MetadataRegistryNotificationParser.NotificationOpType opType) {
+ this.entity = entity;
+ this.classifications = classifications;
+ this.opType = opType;
+ this.entityTypeHandled = MetadataRegistryResourceMapperUtil.isEntityTypeHandled(entity);
+ this.emptyClassifications = CollectionUtils.isEmpty(classifications);
+ }
+
+ public MetadataRegistryEntity getEntity() {
+ return entity;
+ }
+
+ public List<MetadataRegistryNotificationParser.MetadataRegistryClassification> getClassifications() {
+ return classifications;
+ }
+
+ public MetadataRegistryNotificationParser.NotificationOpType getOpType() {
+ return opType;
+ }
+
+ public boolean isEntityTypeHandled() {
+ return entityTypeHandled;
+ }
+
+ public boolean isEmptyClassifications() {
+ return emptyClassifications;
+ }
+
+ public boolean isEntityDeleteOp() {
+ return opType == MetadataRegistryNotificationParser.NotificationOpType.ENTITY_DELETE;
+ }
+
+ public boolean isEntityCreateOp() {
+ return opType == MetadataRegistryNotificationParser.NotificationOpType.ENTITY_CREATE;
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapper.java
new file mode 100644
index 000000000..928ae8659
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapper.java
@@ -0,0 +1,68 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+
+import java.util.Properties;
+
+/**
+ * Maps Metadata Registry entity types to Ranger service resources.
+ */
+public abstract class MetadataRegistryResourceMapper {
+ public static final String TAGSYNC_DEFAULT_CLUSTER_NAME =
+ "ranger.tagsync.metadataregistry.default.cluster.name";
+ public static final String TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX =
+ "ranger.tagsync.metadataregistry.";
+ public static final String TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX =
+ ".ranger.service";
+ public static final String TAGSYNC_CLUSTER_IDENTIFIER = ".instance.";
+ public static final String TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR = "_";
+
+ protected final String componentName;
+ protected final String[] supportedEntityTypes;
+
+ protected Properties properties;
+ protected String defaultClusterName;
+
+ protected MetadataRegistryResourceMapper(String componentName, String[] supportedEntityTypes) {
+ this.componentName = componentName;
+ this.supportedEntityTypes = supportedEntityTypes;
+ }
+
+ public final String getComponentName() {
+ return componentName;
+ }
+
+ public final String[] getSupportedEntityTypes() {
+ return supportedEntityTypes;
+ }
+
+ public void initialize(Properties properties) {
+ this.properties = properties;
+ this.defaultClusterName = properties != null ? properties.getProperty(TAGSYNC_DEFAULT_CLUSTER_NAME) : null;
+ }
+
+ public abstract RangerServiceResource buildResource(MetadataRegistryEntity entity) throws Exception;
+
+ protected String getRangerServiceName(String deployment) {
+ String serviceName = getCustomRangerServiceName(deployment);
+ if (StringUtils.isBlank(serviceName)) {
+ serviceName = deployment + TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR + componentName;
+ }
+ return serviceName;
+ }
+
+ protected String getCustomRangerServiceName(String deployment) {
+ if (properties == null || StringUtils.isBlank(deployment)) {
+ return null;
+ }
+ String propName = TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX + componentName
+ + TAGSYNC_CLUSTER_IDENTIFIER + deployment
+ + TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX;
+ return properties.getProperty(propName);
+ }
+
+ protected void throwExceptionWithMessage(String message) throws Exception {
+ throw new Exception(message);
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapperUtil.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapperUtil.java
new file mode 100644
index 000000000..ef6d516fc
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryResourceMapperUtil.java
@@ -0,0 +1,93 @@
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.ranger.plugin.model.RangerServiceResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public final class MetadataRegistryResourceMapperUtil {
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataRegistryResourceMapperUtil.class);
+
+ private static final Map<String, MetadataRegistryResourceMapper> resourceMappers = new HashMap<>();
+
+ private MetadataRegistryResourceMapperUtil() {}
+
+ public static boolean isEntityTypeHandled(String entityType) {
+ return resourceMappers.containsKey(entityType);
+ }
+
+ public static boolean isEntityTypeHandled(MetadataRegistryEntity entity) {
+ if (entity == null) {
+ return false;
+ }
+ if (MetadataRegistryHiveResourceMapper.isColumnFieldPath(entity.getFieldPath())) {
+ return isEntityTypeHandled(MetadataRegistryHiveResourceMapper.ENTITY_TYPE_HIVE_TABLE);
+ }
+ return isEntityTypeHandled(entity.getEntityType());
+ }
+
+ public static RangerServiceResource getRangerServiceResource(MetadataRegistryEntity entity) {
+ if (entity == null) {
+ return null;
+ }
+ MetadataRegistryResourceMapper mapper = resolveMapper(entity);
+ if (mapper == null) {
+ return null;
+ }
+ try {
+ return mapper.buildResource(entity);
+ } catch (Exception exception) {
+ LOG.error("Could not build Ranger service resource for urn={}", entity.getUrn(), exception);
+ return null;
+ }
+ }
+
+ public static boolean initializeResourceMappers(Properties properties) {
+ String customMapperNames = properties == null
+ ? null
+ : properties.getProperty("ranger.tagsync.metadataregistry.custom.resource.mappers");
+
+ boolean ret = true;
+ List<String> mapperNames = new ArrayList<>();
+ mapperNames.add(MetadataRegistryHiveResourceMapper.class.getName());
+
+ if (customMapperNames != null && !customMapperNames.isBlank()) {
+ for (String mapperName : customMapperNames.split(",")) {
+ mapperNames.add(mapperName.trim());
+ }
+ }
+
+ resourceMappers.clear();
+
+ for (String mapperName : mapperNames) {
+ if (mapperName.isBlank()) {
+ continue;
+ }
+ try {
+ Class<?> clazz = Class.forName(mapperName);
+ MetadataRegistryResourceMapper mapper = (MetadataRegistryResourceMapper) clazz.getDeclaredConstructor().newInstance();
+ mapper.initialize(properties);
+ for (String entityType : mapper.getSupportedEntityTypes()) {
+ resourceMappers.put(entityType, mapper);
+ }
+ } catch (Exception exception) {
+ LOG.error("Failed to create MetadataRegistryResourceMapper: {}", mapperName, exception);
+ ret = false;
+ }
+ }
+
+ return ret;
+ }
+
+ private static MetadataRegistryResourceMapper resolveMapper(MetadataRegistryEntity entity) {
+ if (MetadataRegistryHiveResourceMapper.isColumnFieldPath(entity.getFieldPath())) {
+ return resourceMappers.get(MetadataRegistryHiveResourceMapper.ENTITY_TYPE_HIVE_TABLE);
+ }
+ return resourceMappers.get(entity.getEntityType());
+ }
+}
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryTagSource.java
new file mode 100644
index 000000000..192fd198a
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/metadataregistry/MetadataRegistryTagSource.java
@@ -0,0 +1,243 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.tagsync.source.metadataregistry;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.ranger.authorization.utils.JsonUtils;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Kafka tag source for Metadata Registry {@code udf.governance} notifications.
+ */
+public class MetadataRegistryTagSource extends AbstractTagSource {
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataRegistryTagSource.class);
+
+ public static final String TAGSYNC_METADATAREGISTRY_KAFKA_BOOTSTRAP =
+ "ranger.tagsync.source.metadataregistry.kafka.bootstrap.servers";
+ public static final String TAGSYNC_METADATAREGISTRY_KAFKA_TOPIC =
+ "ranger.tagsync.source.metadataregistry.kafka.topic";
+ public static final String TAGSYNC_METADATAREGISTRY_KAFKA_GROUP =
+ "ranger.tagsync.source.metadataregistry.kafka.group.id";
+
+ private static final String DEFAULT_TOPIC = "udf.governance";
+ private static final String DEFAULT_GROUP = "ranger_metadataregistry_consumer";
+ private static final int MAX_WAIT_TIME_MS = 1000;
+
+ private int maxBatchSize;
+ private ConsumerRunnable consumerTask;
+ private Thread consumerThread;
+ private Properties tagSyncProperties;
+
+ @Override
+ public boolean initialize(Properties properties) {
+ LOG.debug("==> MetadataRegistryTagSource.initialize()");
+
+ tagSyncProperties = properties;
+ boolean ret = MetadataRegistryResourceMapperUtil.initializeResourceMappers(properties);
+ if (ret) {
+ if (StringUtils.isBlank(properties.getProperty(TAGSYNC_METADATAREGISTRY_KAFKA_BOOTSTRAP))) {
+ LOG.error("missing value for mandatory property '{}'", TAGSYNC_METADATAREGISTRY_KAFKA_BOOTSTRAP);
+ ret = false;
+ } else {
+ maxBatchSize = TagSyncConfig.getSinkMaxBatchSize(properties);
+ consumerTask = new ConsumerRunnable(properties);
+ }
+ }
+
+ LOG.debug("<== MetadataRegistryTagSource.initialize(), result={}", ret);
+ return ret;
+ }
+
+ @Override
+ public boolean start() {
+ LOG.debug("==> MetadataRegistryTagSource.start()");
+
+ boolean ret = false;
+ if (consumerTask != null) {
+ consumerThread = new Thread(consumerTask);
+ consumerThread.setDaemon(true);
+ consumerThread.start();
+ ret = true;
+ } else {
+ LOG.error("No Metadata Registry consumer task configured");
+ }
+
+ LOG.debug("<== MetadataRegistryTagSource.start(): ret={}", ret);
+ return ret;
+ }
+
+ @Override
+ public void stop() {
+ if (consumerThread != null && consumerThread.isAlive()) {
+ consumerThread.interrupt();
+ }
+ if (consumerTask != null) {
+ consumerTask.close();
+ }
+ }
+
+ private class ConsumerRunnable implements Runnable {
+ private final KafkaConsumer<String, String> consumer;
+ private final List<MetadataRegistryEntityWithTags> entitiesWithTags = new ArrayList<>();
+ private final List<ConsumerRecord<String, String>> messages = new ArrayList<>();
+ private ConsumerRecord<String, String> lastUnhandledMessage;
+ private boolean isHandlingDeleteOps;
+
+ private ConsumerRunnable(Properties properties) {
+ Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ properties.getProperty(TAGSYNC_METADATAREGISTRY_KAFKA_BOOTSTRAP));
+ consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG,
+ properties.getProperty(TAGSYNC_METADATAREGISTRY_KAFKA_GROUP, DEFAULT_GROUP));
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+ consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ this.consumer = new KafkaConsumer<>(consumerProps);
+ String topic = properties.getProperty(TAGSYNC_METADATAREGISTRY_KAFKA_TOPIC, DEFAULT_TOPIC);
+ consumer.subscribe(Collections.singletonList(topic));
+ }
+
+ private void close() {
+ consumer.close();
+ }
+
+ @Override
+ public void run() {
+ LOG.debug("==> MetadataRegistryTagSource.ConsumerRunnable.run()");
+
+ while (!Thread.currentThread().isInterrupted()) {
+ if (!TagSyncConfig.isTagSyncServiceActive()) {
+ sleepPassive();
+ continue;
+ }
+ try {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(MAX_WAIT_TIME_MS));
+ if (records.isEmpty()) {
+ if (CollectionUtils.isNotEmpty(entitiesWithTags)) {
+ buildAndUploadServiceTags();
+ }
+ } else {
+ for (ConsumerRecord<String, String> record : records) {
+ MetadataRegistryNotificationWrapper wrapper =
+ MetadataRegistryNotificationParser.parse(record.value());
+ if (wrapper == null) {
+ lastUnhandledMessage = record;
+ continue;
+ }
+ if (MetadataRegistryNotificationMapper.isNotificationHandled(wrapper)) {
+ if ((wrapper.isEntityDeleteOp() && !isHandlingDeleteOps)
+ || (!wrapper.isEntityDeleteOp() && isHandlingDeleteOps)) {
+ if (CollectionUtils.isNotEmpty(entitiesWithTags)) {
+ buildAndUploadServiceTags();
+ }
+ isHandlingDeleteOps = wrapper.isEntityDeleteOp();
+ }
+ entitiesWithTags.add(new MetadataRegistryEntityWithTags(wrapper));
+ messages.add(record);
+ } else {
+ MetadataRegistryNotificationMapper.logUnhandledNotification(wrapper);
+ lastUnhandledMessage = record;
+ }
+ }
+ if (CollectionUtils.isNotEmpty(entitiesWithTags) && entitiesWithTags.size() >= maxBatchSize) {
+ buildAndUploadServiceTags();
+ }
+ }
+ if (lastUnhandledMessage != null) {
+ commitToKafka(lastUnhandledMessage);
+ lastUnhandledMessage = null;
+ }
+ } catch (Exception exception) {
+ LOG.error("Metadata Registry tag source consumer error", exception);
+ sleepQuietly(100);
+ }
+ }
+ }
+
+ private void buildAndUploadServiceTags() throws Exception {
+ if (CollectionUtils.isEmpty(entitiesWithTags) || CollectionUtils.isEmpty(messages)) {
+ return;
+ }
+ Map<String, ServiceTags> serviceTagsMap =
+ MetadataRegistryNotificationMapper.processEntities(entitiesWithTags);
+ if (MapUtils.isNotEmpty(serviceTagsMap)) {
+ for (Map.Entry<String, ServiceTags> entry : serviceTagsMap.entrySet()) {
+ if (isHandlingDeleteOps) {
+ entry.getValue().setOp(ServiceTags.OP_DELETE);
+ entry.getValue().setTagDefinitions(Collections.emptyMap());
+ entry.getValue().setTags(Collections.emptyMap());
+ } else {
+ entry.getValue().setOp(ServiceTags.OP_ADD_OR_UPDATE);
+ }
+ LOG.debug("Metadata Registry serviceTags={}", JsonUtils.objectToJson(entry.getValue()));
+ updateSink(entry.getValue());
+ }
+ }
+ ConsumerRecord<String, String> latest = messages.get(messages.size() - 1);
+ commitToKafka(latest);
+ entitiesWithTags.clear();
+ messages.clear();
+ }
+
+ private void commitToKafka(ConsumerRecord<String, String> record) {
+ TopicPartition partition = new TopicPartition(record.topic(), record.partition());
+ consumer.commitSync(Collections.singletonMap(
+ partition,
+ new OffsetAndMetadata(record.offset() + 1)));
+ }
+
+ private void sleepPassive() {
+ try {
+ Thread.sleep(TagSyncConfig.getTagSyncHAPassiveSleepInterval());
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void sleepQuietly(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+}
diff --git a/tagsync/src/main/resources/ranger-tagsync-site.xml b/tagsync/src/main/resources/ranger-tagsync-site.xml
index 672930f21..c52afafe3 100644
--- a/tagsync/src/main/resources/ranger-tagsync-site.xml
+++ b/tagsync/src/main/resources/ranger-tagsync-site.xml
@@ -41,6 +41,26 @@
<name>ranger.tagsync.source.atlasrest</name>
<value>false</value>
</property>
+ <property>
+ <name>ranger.tagsync.source.metadataregistry</name>
+ <value>false</value>
+ </property>
+ <property>
+ <name>ranger.tagsync.source.metadataregistry.kafka.bootstrap.servers</name>
+ <value>localhost:9092</value>
+ </property>
+ <property>
+ <name>ranger.tagsync.source.metadataregistry.kafka.topic</name>
+ <value>udf.governance</value>
+ </property>
+ <property>
+ <name>ranger.tagsync.source.metadataregistry.kafka.group.id</name>
+ <value>ranger_metadataregistry_consumer</value>
+ </property>
+ <property>
+ <name>ranger.tagsync.metadataregistry.custom.resource.mappers</name>
+ <value />
+ </property>
<property>
<name>ranger.tagsync.source.atlasrest.endpoint</name>
<value>http://localhost:21000</value>
diff --git a/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestMetadataRegistryHiveResourceMapper.java b/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestMetadataRegistryHiveResourceMapper.java
new file mode 100644
index 000000000..8788bb4b7
--- /dev/null
+++ b/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestMetadataRegistryHiveResourceMapper.java
@@ -0,0 +1,73 @@
+package org.apache.ranger.tagsync.process;
+
+import org.apache.ranger.plugin.model.RangerServiceResource;
+import org.apache.ranger.tagsync.source.metadataregistry.MetadataRegistryEntity;
+import org.apache.ranger.tagsync.source.metadataregistry.MetadataRegistryHiveResourceMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestMetadataRegistryHiveResourceMapper {
+ private final MetadataRegistryHiveResourceMapper mapper = new MetadataRegistryHiveResourceMapper();
+
+ @Test
+ public void testHiveTableUrn() throws Exception {
+ MetadataRegistryEntity entity = new MetadataRegistryEntity(
+ "urn:hive:table::hms:prod:sales:orders",
+ 42L,
+ MetadataRegistryHiveResourceMapper.ENTITY_TYPE_HIVE_TABLE,
+ "",
+ "orders",
+ "Orders",
+ "prod",
+ "hms");
+
+ RangerServiceResource resource = mapper.buildResource(entity);
+
+ Assertions.assertEquals("prod_hive", resource.getServiceName());
+ Assertions.assertEquals(
+ "sales",
+ resource.getResourceElements().get("database").getValues().get(0));
+ Assertions.assertEquals(
+ "orders",
+ resource.getResourceElements().get("table").getValues().get(0));
+ }
+
+ @Test
+ public void testHiveDatabaseUrn() throws Exception {
+ MetadataRegistryEntity entity = new MetadataRegistryEntity(
+ "urn:hive:database::hms:prod:sales",
+ 7L,
+ MetadataRegistryHiveResourceMapper.ENTITY_TYPE_HIVE_DATABASE,
+ "",
+ "sales",
+ "Sales",
+ "prod",
+ "hms");
+
+ RangerServiceResource resource = mapper.buildResource(entity);
+
+ Assertions.assertEquals("prod_hive", resource.getServiceName());
+ Assertions.assertEquals(
+ "sales",
+ resource.getResourceElements().get("database").getValues().get(0));
+ }
+
+ @Test
+ public void testHiveColumnFieldPath() throws Exception {
+ MetadataRegistryEntity entity = new MetadataRegistryEntity(
+ "urn:hive:table::hms:prod:sales:orders",
+ 42L,
+ MetadataRegistryHiveResourceMapper.ENTITY_TYPE_HIVE_TABLE,
+ "columns.email",
+ "orders",
+ "Orders",
+ "prod",
+ "hms");
+
+ RangerServiceResource resource = mapper.buildResource(entity);
+
+ Assertions.assertEquals(
+ "email",
+ resource.getResourceElements().get("column").getValues().get(0));
+ }
+}