Skip to content

A
atlas
Commit 570f2423 by Madhan Neethiraj
Options

ATLAS-2656: updated entity notifications to replace AtlasEntity with AtlasEntityHeader

parent 5fed48d9
Showing with 192 additions and 213 deletions
addons/models/1000-Hadoop/1020-fs_model.json
......@@ -94,6 +94,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": false,
"includeInNotification": true,
"isUnique": false
},
{
......@@ -167,6 +168,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": true,
"includeInNotification": true,
"isUnique": false
},
{
......@@ -191,6 +193,7 @@
"cardinality": "SINGLE",
"isIndexable": false,
"isOptional": true,
"includeInNotification": true,
"isUnique": false
}
]
......
addons/models/1000-Hadoop/1030-hive_model.json
......@@ -152,6 +152,7 @@
"cardinality": "SINGLE",
"isIndexable": false,
"isOptional": true,
"includeInNotification": true,
"isUnique": false
},
{
......@@ -418,6 +419,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": false,
"includeInNotification": true,
"isUnique": false
},
{
......
addons/models/1000-Hadoop/1060-hbase_model.json
......@@ -16,6 +16,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": false,
"includeInNotification": true,
"isUnique": false
},
{
......
addons/models/1000-Hadoop/1070-kafka_model.json
......@@ -16,6 +16,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": false,
"isInNotification": true,
"isUnique": false
},
{
......
addons/models/1000-Hadoop/1080-storm_model.json
......@@ -48,6 +48,7 @@
"cardinality": "SINGLE",
"isIndexable": true,
"isOptional": true,
"includeInNotification": true,
"isUnique": false
},
{
......
intg/src/main/java/org/apache/atlas/model/notification/EntityNotification.java
......@@ -20,14 +20,17 @@ package org.apache.atlas.model.notification;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.atlas.model.instance.AtlasEntityHeader;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
import java.io.Serializable;
import java.util.Objects;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
import static org.apache.atlas.model.notification.EntityNotification.EntityNotificationType.ENTITY_NOTIFICATION_V2;
/**
* Base type of hook message.
......@@ -83,4 +86,86 @@ public class EntityNotification implements Serializable {
return sb;
}
/**
* Entity v2 notification
*/
@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown=true)
@XmlRootElement
@XmlAccessorType(XmlAccessType.PROPERTY)
public static class EntityNotificationV2 extends EntityNotification implements Serializable {
private static final long serialVersionUID = 1L;
public enum OperationType {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE,
CLASSIFICATION_ADD, CLASSIFICATION_DELETE, CLASSIFICATION_UPDATE
}
private AtlasEntityHeader entity;
private OperationType operationType;
public EntityNotificationV2() {
super(ENTITY_NOTIFICATION_V2);
}
public EntityNotificationV2(AtlasEntityHeader entity, OperationType operationType) {
super(ENTITY_NOTIFICATION_V2);
setEntity(entity);
setOperationType(operationType);
}
public AtlasEntityHeader getEntity() {
return entity;
}
public void setEntity(AtlasEntityHeader entity) {
this.entity = entity;
}
public OperationType getOperationType() {
return operationType;
}
public void setOperationType(OperationType operationType) {
this.operationType = operationType;
}
@Override
public boolean equals(Object o) {
if (this == o) { return true; }
if (o == null || getClass() != o.getClass()) { return false; }
EntityNotificationV2 that = (EntityNotificationV2) o;
return Objects.equals(type, that.type) &&
Objects.equals(entity, that.entity) &&
operationType == that.operationType;
}
@Override
public int hashCode() {
return Objects.hash(type, entity, operationType);
}
@Override
public StringBuilder toString(StringBuilder sb) {
if (sb == null) {
sb = new StringBuilder();
}
sb.append("EntityNotificationV1{");
super.toString(sb);
sb.append(", entity=");
if (entity != null) {
entity.toString(sb);
} else {
sb.append(entity);
}
sb.append(", operationType=").append(operationType);
sb.append("}");
return sb;
}
}
}
intg/src/main/java/org/apache/atlas/model/typedef/AtlasStructDef.java
......@@ -272,6 +272,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
private int valuesMaxCount;
private boolean isUnique;
private boolean isIndexable;
private boolean includeInNotification;
private String defaultValue;
private String description;
......@@ -280,17 +281,17 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
public AtlasAttributeDef() { this(null, null); }
public AtlasAttributeDef(String name, String typeName) {
this(name, typeName, false, Cardinality.SINGLE, COUNT_NOT_SET, COUNT_NOT_SET, false, false, null);
this(name, typeName, false, Cardinality.SINGLE, COUNT_NOT_SET, COUNT_NOT_SET, false, false, false, null);
}
public AtlasAttributeDef(String name, String typeName, boolean isOptional, Cardinality cardinality,
int valuesMinCount, int valuesMaxCount, boolean isUnique, boolean isIndexable, List<AtlasConstraintDef> constraints) {
this(name, typeName, isOptional, cardinality, valuesMinCount, valuesMaxCount, isUnique, isIndexable, null, constraints, null);
int valuesMinCount, int valuesMaxCount, boolean isUnique, boolean isIndexable, boolean includeInNotification, List<AtlasConstraintDef> constraints) {
this(name, typeName, isOptional, cardinality, valuesMinCount, valuesMaxCount, isUnique, isIndexable, includeInNotification, null, constraints, null);
}
public AtlasAttributeDef(String name, String typeName, boolean isOptional, Cardinality cardinality,
int valuesMinCount, int valuesMaxCount, boolean isUnique, boolean isIndexable, String defaultValue,
int valuesMinCount, int valuesMaxCount, boolean isUnique, boolean isIndexable, boolean includeInNotification, String defaultValue,
List<AtlasConstraintDef> constraints, String description) {
setName(name);
setTypeName(typeName);
......@@ -300,6 +301,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
setValuesMaxCount(valuesMaxCount);
setIsUnique(isUnique);
setIsIndexable(isIndexable);
setIncludeInNotification(includeInNotification);
setDefaultValue(defaultValue);
setConstraints(constraints);
setDescription(description);
......@@ -315,6 +317,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
setValuesMaxCount(other.getValuesMaxCount());
setIsUnique(other.getIsUnique());
setIsIndexable(other.getIsIndexable());
setIncludeInNotification(other.getIncludeInNotification());
setDefaultValue(other.getDefaultValue());
setConstraints(other.getConstraints());
setDescription((other.getDescription()));
......@@ -379,6 +382,10 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
return isIndexable;
}
public boolean getIncludeInNotification() { return includeInNotification; }
public void setIncludeInNotification(Boolean isInNotification) { this.includeInNotification = isInNotification == null ? Boolean.FALSE : isInNotification; }
public String getDefaultValue(){
return defaultValue;
}
......@@ -440,6 +447,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
sb.append(", valuesMaxCount=").append(valuesMaxCount);
sb.append(", isUnique=").append(isUnique);
sb.append(", isIndexable=").append(isIndexable);
sb.append(", includeInNotification=").append(includeInNotification);
sb.append(", defaultValue=").append(defaultValue);
sb.append(", constraints=[");
if (CollectionUtils.isNotEmpty(constraints)) {
......@@ -468,6 +476,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
valuesMaxCount == that.valuesMaxCount &&
isUnique == that.isUnique &&
isIndexable == that.isIndexable &&
includeInNotification == that.includeInNotification &&
Objects.equals(name, that.name) &&
Objects.equals(typeName, that.typeName) &&
cardinality == that.cardinality &&
......@@ -478,7 +487,7 @@ public class AtlasStructDef extends AtlasBaseTypeDef implements Serializable {
@Override
public int hashCode() {
return Objects.hash(name, typeName, isOptional, cardinality, valuesMinCount, valuesMaxCount, isUnique, isIndexable, defaultValue, constraints, description);
return Objects.hash(name, typeName, isOptional, cardinality, valuesMinCount, valuesMaxCount, isUnique, isIndexable, includeInNotification, defaultValue, constraints, description);
}
@Override
......
intg/src/main/java/org/apache/atlas/type/AtlasTypeUtil.java
......@@ -143,35 +143,35 @@ public class AtlasTypeUtil {
public static AtlasAttributeDef createOptionalAttrDef(String name, AtlasType dataType) {
return new AtlasAttributeDef(name, dataType.getTypeName(), true,
Cardinality.SINGLE, 0, 1,
false, false,
false, false, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createOptionalAttrDef(String name, String dataType) {
return new AtlasAttributeDef(name, dataType, true,
Cardinality.SINGLE, 0, 1,
false, false,
false, false, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createRequiredAttrDef(String name, String dataType) {
return new AtlasAttributeDef(name, dataType, false,
Cardinality.SINGLE, 1, 1,
false, true,
false, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createListRequiredAttrDef(String name, String dataType) {
return new AtlasAttributeDef(name, dataType, false,
Cardinality.LIST, 1, Integer.MAX_VALUE,
false, true,
false, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createOptionalListAttrDef(String name, String dataType) {
return new AtlasAttributeDef(name, dataType, true,
Cardinality.LIST, 1, Integer.MAX_VALUE,
false, true,
false, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
......@@ -199,21 +199,21 @@ public class AtlasTypeUtil {
public static AtlasAttributeDef createUniqueRequiredAttrDef(String name, AtlasType dataType) {
return new AtlasAttributeDef(name, dataType.getTypeName(), false,
Cardinality.SINGLE, 1, 1,
true, true,
true, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createUniqueRequiredAttrDef(String name, String typeName) {
return new AtlasAttributeDef(name, typeName, false,
Cardinality.SINGLE, 1, 1,
true, true,
true, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
public static AtlasAttributeDef createRequiredAttrDef(String name, AtlasType dataType) {
return new AtlasAttributeDef(name, dataType.getTypeName(), false,
Cardinality.SINGLE, 1, 1,
false, true,
false, true, false,
Collections.<AtlasConstraintDef>emptyList());
}
......
intg/src/main/java/org/apache/atlas/utils/AtlasJson.java
......@@ -27,6 +27,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.atlas.model.discovery.AtlasSearchResult.AtlasFullTextResult;
import org.apache.atlas.model.notification.EntityNotification;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationType;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2;
import org.apache.atlas.model.notification.HookNotification;
import org.apache.atlas.model.notification.HookNotification.HookNotificationType;
import org.apache.atlas.model.notification.HookNotification.EntityCreateRequestV2;
......@@ -39,7 +40,6 @@ import org.apache.atlas.v1.model.instance.Id;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.v1.model.notification.EntityNotificationV1;
import org.apache.atlas.v1.model.notification.EntityNotificationV2;
import org.apache.atlas.v1.model.notification.HookNotificationV1.*;
import org.apache.commons.collections.MapUtils;
import org.slf4j.Logger;
......
intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotificationV2.java deleted 100644 → 0
/**
* 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.atlas.v1.model.notification;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.notification.EntityNotification;
import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
import java.io.Serializable;
import java.util.List;
import java.util.Objects;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
import static org.apache.atlas.model.notification.EntityNotification.EntityNotificationType.ENTITY_NOTIFICATION_V2;
/**
* Entity v2 notification
*/
@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown=true)
@XmlRootElement
@XmlAccessorType(XmlAccessType.PROPERTY)
public class EntityNotificationV2 extends EntityNotification implements Serializable {
private static final long serialVersionUID = 1L;
public enum OperationType {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE,
CLASSIFICATION_ADD, CLASSIFICATION_DELETE, CLASSIFICATION_UPDATE
}
private AtlasEntity entity;
private OperationType operationType;
private List<AtlasClassification> classifications;
public EntityNotificationV2() { }
public EntityNotificationV2(AtlasEntity entity, OperationType operationType, List<AtlasClassification> classifications) {
setEntity(entity);
setOperationType(operationType);
setClassifications(classifications);
setType(ENTITY_NOTIFICATION_V2);
}
public AtlasEntity getEntity() {
return entity;
}
public void setEntity(AtlasEntity entity) {
this.entity = entity;
}
public OperationType getOperationType() {
return operationType;
}
public void setOperationType(OperationType operationType) {
this.operationType = operationType;
}
public List<AtlasClassification> getClassifications() {
return classifications;
}
public void setClassifications(List<AtlasClassification> classifications) {
this.classifications = classifications;
}
@Override
public boolean equals(Object o) {
if (this == o) { return true; }
if (o == null || getClass() != o.getClass()) { return false; }
EntityNotificationV2 that = (EntityNotificationV2) o;
return Objects.equals(entity, that.entity) &&
operationType == that.operationType &&
Objects.equals(classifications, that.classifications);
}
@Override
public int hashCode() {
return Objects.hash(entity, operationType, classifications);
}
@Override
public StringBuilder toString(StringBuilder sb) {
if (sb == null) {
sb = new StringBuilder();
}
sb.append("EntityNotificationV1{");
super.toString(sb);
sb.append(", entity=");
if (entity != null) {
entity.toString(sb);
} else {
sb.append(entity);
}
sb.append(", operationType=").append(operationType);
sb.append(", classifications=[");
AtlasBaseTypeDef.dumpObjects(classifications, sb);
sb.append("]");
sb.append("}");
return sb;
}
}
\ No newline at end of file
repository/src/main/java/org/apache/atlas/repository/converters/AtlasInstanceConverter.java
......@@ -56,13 +56,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_ADD;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_DELETE;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_UPDATE;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditAction.CLASSIFICATION_DELETE;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditAction.CLASSIFICATION_UPDATE;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.CLASSIFICATION_ADD;
@Singleton
@Component
public class AtlasInstanceConverter {
......
repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasStructDefStoreV1.java
......@@ -480,6 +480,7 @@ public class AtlasStructDefStoreV1 extends AtlasAbstractDefStoreV1<AtlasStructDe
attribInfo.put("dataType", attributeDef.getTypeName());
attribInfo.put("isUnique", attributeDef.getIsUnique());
attribInfo.put("isIndexable", attributeDef.getIsIndexable());
attribInfo.put("includeInNotification", attributeDef.getIncludeInNotification());
attribInfo.put("isComposite", attribute.isOwnedRef());
attribInfo.put("reverseAttributeName", attribute.getInverseRefAttributeName());
attribInfo.put("defaultValue", attributeDef.getDefaultValue());
......@@ -522,6 +523,7 @@ public class AtlasStructDefStoreV1 extends AtlasAbstractDefStoreV1<AtlasStructDe
ret.setTypeName((String) attribInfo.get("dataType"));
ret.setIsUnique((Boolean) attribInfo.get("isUnique"));
ret.setIsIndexable((Boolean) attribInfo.get("isIndexable"));
ret.setIncludeInNotification((Boolean) attribInfo.get("includeInNotification"));
ret.setDefaultValue((String) attribInfo.get("defaultValue"));
ret.setDescription((String) attribInfo.get("description"));
......
repository/src/main/java/org/apache/atlas/repository/store/graph/v1/EntityGraphRetriever.java
......@@ -118,12 +118,12 @@ public final class EntityGraphRetriever {
private static final String TERM_RELATION_NAME = "AtlasGlossarySemanticAssignment";
private static final String GLOSSARY_TERM_DISPLAY_NAME_ATTR = "AtlasGlossaryTerm.displayName";
private final String NAME = "name";
private final String DISPLAY_NAME = "displayName";
private final String DESCRIPTION = "description";
private final String OWNER = "owner";
private final String CREATE_TIME = "createTime";
private final String QUALIFIED_NAME = "qualifiedName";
public static final String NAME = "name";
public static final String DISPLAY_NAME = "displayName";
public static final String DESCRIPTION = "description";
public static final String OWNER = "owner";
public static final String CREATE_TIME = "createTime";
public static final String QUALIFIED_NAME = "qualifiedName";
private static final TypeReference<List<TimeBoundary>> TIME_BOUNDARIES_LIST_TYPE = new TypeReference<List<TimeBoundary>>() {};
private static final GraphHelper graphHelper = GraphHelper.getInstance();
......
repository/src/test/java/org/apache/atlas/repository/store/graph/AtlasTypeDefGraphStoreTest.java
......@@ -570,7 +570,7 @@ public class AtlasTypeDefGraphStoreTest {
AtlasClassificationDef aTag = new AtlasClassificationDef("testTag");
AtlasAttributeDef attributeDef = new AtlasAttributeDef("testAttribute", "string", true,
AtlasAttributeDef.Cardinality.SINGLE, 0, 1,
false, true,
false, true, false,
Collections.<AtlasStructDef.AtlasConstraintDef>emptyList());
aTag.addAttribute(attributeDef);
......@@ -592,7 +592,7 @@ public class AtlasTypeDefGraphStoreTest {
aTag = new AtlasClassificationDef("testTag");
attributeDef = new AtlasAttributeDef("testAttribute", "int", true,
AtlasAttributeDef.Cardinality.SINGLE, 0, 1,
false, true,
false, true, false,
Collections.<AtlasStructDef.AtlasConstraintDef>emptyList());
aTag.addAttribute(attributeDef);
typesDef.setClassificationDefs(Arrays.asList(aTag));
......
repository/src/test/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityStoreV1Test.java
......@@ -1007,7 +1007,7 @@ public class AtlasEntityStoreV1Test {
private AtlasClassificationDef getTagWithName(AtlasTypesDef typesDef, String tagName, String attributeType) {
AtlasClassificationDef aTag = new AtlasClassificationDef(tagName);
AtlasAttributeDef attributeDef = new AtlasAttributeDef("testAttribute", attributeType, true,
AtlasAttributeDef.Cardinality.SINGLE, 0, 1, false, true,
AtlasAttributeDef.Cardinality.SINGLE, 0, 1, false, true, false,
Collections.emptyList());
aTag.addAttribute(attributeDef);
......
webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
......@@ -23,10 +23,13 @@ import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.listener.EntityChangeListenerV2;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntityHeader;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2.OperationType;
import org.apache.atlas.type.AtlasClassificationType;
import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.v1.model.notification.EntityNotificationV2;
import org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.configuration.Configuration;
......@@ -34,26 +37,25 @@ import org.springframework.stereotype.Component;
import javax.inject.Inject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.apache.atlas.notification.NotificationEntityChangeListener.ATLAS_ENTITY_NOTIFICATION_PROPERTY;
import static org.apache.atlas.notification.NotificationInterface.NotificationType.ENTITIES;
import static org.apache.atlas.repository.graph.GraphHelper.isInternalType;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.*;
import static org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2.OperationType.*;
import static org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever.CREATE_TIME;
import static org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever.DESCRIPTION;
import static org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever.NAME;
import static org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever.OWNER;
import static org.apache.atlas.repository.store.graph.v1.EntityGraphRetriever.QUALIFIED_NAME;
@Component
public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
private final AtlasTypeRegistry typeRegistry;
private final NotificationInterface notificationInterface;
private final Configuration configuration;
private final Map<String, List<String>> notificationAttributesCache = new HashMap<>();
@Inject
public EntityNotificationListenerV2(AtlasTypeRegistry typeRegistry,
......@@ -61,7 +63,6 @@ public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
Configuration configuration) {
this.typeRegistry = typeRegistry;
this.notificationInterface = notificationInterface;
this.configuration = configuration;
}
@Override
......@@ -109,9 +110,7 @@ public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
continue;
}
filterNotificationAttributes(entity);
messages.add(new EntityNotificationV2(entity, operationType, getAllClassifications(entity)));
messages.add(new EntityNotificationV2(toNotificationHeader(entity), operationType));
}
if (!messages.isEmpty()) {
......@@ -123,14 +122,60 @@ public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
}
}
private List<AtlasClassification> getAllClassifications(AtlasEntity entity) {
List<AtlasClassification> ret = getAllClassifications(entity.getClassifications(), typeRegistry);
private AtlasEntityHeader toNotificationHeader(AtlasEntity entity) {
AtlasEntityHeader ret = new AtlasEntityHeader(entity.getTypeName(), entity.getGuid(), new HashMap<>());
Object name = entity.getAttribute(NAME);
Object displayText = name != null ? name : entity.getAttribute(QUALIFIED_NAME);
ret.setGuid(entity.getGuid());
ret.setStatus(entity.getStatus());
setAttribute(ret, NAME, name);
setAttribute(ret, DESCRIPTION, entity.getAttribute(DESCRIPTION));
setAttribute(ret, OWNER, entity.getAttribute(OWNER));
setAttribute(ret, CREATE_TIME, entity.getAttribute(CREATE_TIME));
if (displayText != null) {
ret.setDisplayText(displayText.toString());
}
AtlasEntityType entityType = typeRegistry.getEntityTypeByName(entity.getTypeName());
if (entityType != null) {
for (AtlasAttribute attribute : entityType.getAllAttributes().values()) {
if (attribute.getAttributeDef().getIsUnique() || attribute.getAttributeDef().getIncludeInNotification()) {
Object attrValue = entity.getAttribute(attribute.getName());
if (attrValue != null) {
ret.setAttribute(attribute.getName(), attrValue);
}
}
}
if (CollectionUtils.isNotEmpty(entity.getClassifications())) {
List<AtlasClassification> classifications = new ArrayList<>(entity.getClassifications().size());
List<String> classificationNames = new ArrayList<>(entity.getClassifications().size());
for (AtlasClassification classification : getAllClassifications(entity.getClassifications())) {
classifications.add(classification);
classificationNames.add(classification.getTypeName());
}
ret.setClassifications(classifications);
ret.setClassificationNames(classificationNames);
}
}
return ret;
}
private static List<AtlasClassification> getAllClassifications(List<AtlasClassification> classifications, AtlasTypeRegistry typeRegistry) {
List<AtlasClassification> ret = new LinkedList<>();
private void setAttribute(AtlasEntityHeader entity, String attrName, Object attrValue) {
if (attrValue != null) {
entity.setAttribute(attrName, attrValue);
}
}
private List<AtlasClassification> getAllClassifications(List<AtlasClassification> classifications) {
List<AtlasClassification> ret = new ArrayList<>();
if (CollectionUtils.isNotEmpty(classifications)) {
for (AtlasClassification classification : classifications) {
......@@ -172,36 +217,4 @@ public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
return ret;
}
private void filterNotificationAttributes(AtlasEntity entity) {
Map<String, Object> attributesMap = entity.getAttributes();
List<String> notificationAttrs = getNotificationAttributes(entity.getTypeName());
if (MapUtils.isNotEmpty(attributesMap) && CollectionUtils.isNotEmpty(notificationAttrs)) {
Collection<String> attributesToRemove = CollectionUtils.subtract(attributesMap.keySet(), notificationAttrs);
for (String attributeToRemove : attributesToRemove) {
attributesMap.remove(attributeToRemove);
}
}
}
private List<String> getNotificationAttributes(String entityType) {
List<String> ret = null;
if (notificationAttributesCache.containsKey(entityType)) {
ret = notificationAttributesCache.get(entityType);
} else if (configuration != null) {
String attributesToIncludeKey = ATLAS_ENTITY_NOTIFICATION_PROPERTY + "." + entityType + "." + "attributes.include";
String[] notificationAttributes = configuration.getStringArray(attributesToIncludeKey);
if (notificationAttributes != null) {
ret = Arrays.asList(notificationAttributes);
}
notificationAttributesCache.put(entityType, ret);
}
return ret;
}
}
\ No newline at end of file
webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
......@@ -379,7 +379,7 @@ public abstract class BaseResourceIT {
AtlasBaseTypeDef.getArrayTypeName(HIVE_TABLE_TYPE_V2),
true,
Cardinality.SET,
0, -1, false, true, Collections.singletonList(isCompositeSourceConstraint))
0, -1, false, true, false, Collections.singletonList(isCompositeSourceConstraint))
);
AtlasEntityDef columnClsDef = AtlasTypeUtil
......@@ -409,7 +409,7 @@ public abstract class BaseResourceIT {
DATABASE_TYPE_V2,
true,
Cardinality.SINGLE,
0, 1, false, true, Collections.singletonList(isCompositeTargetConstraint)),
0, 1, false, true, false, Collections.singletonList(isCompositeTargetConstraint)),
//some tests don't set the columns field or set it to null...
AtlasTypeUtil.createOptionalAttrDef("columns", AtlasBaseTypeDef.getArrayTypeName(COLUMN_TYPE_V2)),
......
webapp/src/test/java/org/apache/atlas/web/integration/TypedefsJerseyResourceIT.java
......@@ -314,7 +314,7 @@ public class TypedefsJerseyResourceIT extends BaseResourceIT {
new AtlasAttributeDef("database", "database",
false,
Cardinality.SINGLE, 1, 1,
true, true,
true, true, false,
Collections.<AtlasConstraintDef>emptyList()));
atlasTypesDef.getEntityDefs().add(tableTypeDefinition);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment