ATLAS-1980: Add relationship instance validation to check end point type

intg/src/main/java/org/apache/atlas/AtlasErrorCode.java

@@ -89,7 +89,8 @@ public enum AtlasErrorCode {
     RELATIONSHIPDEF_END2_NAME_INVALID(400, "ATLAS-400-00-042", "{0}: invalid end2 name. Name must not contain query keywords"),
     RELATIONSHIPDEF_NOT_DEFINED(400, "ATLAS-400-00-043", "No relationshipDef defined between {0} and {1} on attribute: {2}"),
     RELATIONSHIPDEF_INVALID(400, "ATLAS-400-00-044", "Invalid relationshipDef: {0}"),
-    // All Not found enums go here
+    RELATIONSHIP_INVALID_ENDTYPE(400, "ATLAS-400-00-045", "Invalid entity-type for relationship attribute ‘{0}’: entity specified (guid={1}) is of type ‘{2}’, but expected type is ‘{3}’"),
+     // All Not found enums go here
     TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-001", "Given typename {0} was invalid"),
     TYPE_GUID_NOT_FOUND(404, "ATLAS-404-00-002", "Given type guid {0} was invalid"),
     EMPTY_RESULTS(404, "ATLAS-404-00-004", "No result found for {0}"),

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasRelationshipStoreV1.java

@@ -23,22 +23,19 @@ import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.TypeCategory;
 import org.apache.atlas.model.instance.AtlasObjectId;
 import org.apache.atlas.model.instance.AtlasRelationship;
+import org.apache.atlas.model.typedef.AtlasRelationshipDef;
 import org.apache.atlas.model.typedef.AtlasRelationshipEndDef;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.RepositoryException;
 import org.apache.atlas.repository.graph.GraphHelper;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasVertex;
-import org.apache.atlas.repository.graphdb.GremlinVersion;
 import org.apache.atlas.repository.store.graph.AtlasRelationshipStore;
 import org.apache.atlas.type.AtlasEntityType;
 import org.apache.atlas.type.AtlasRelationshipType;
-import org.apache.atlas.type.AtlasStructType;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
-import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.atlas.type.AtlasTypeUtil;
-import org.apache.atlas.typesystem.exception.EntityNotFoundException;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -47,10 +44,10 @@ import org.springframework.stereotype.Component;
 
 import javax.inject.Inject;
 import java.util.ArrayList;
-import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 
@@ -237,28 +234,55 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
                                          relationshipType.getEnd1Type().getTypeName(), end2TypeName);
         }
 
-        validateEnd(relationship.getEnd1());
-
-        validateEnd(relationship.getEnd2());
+        validateEnds(relationship);
 
         validateAndNormalize(relationship);
     }
 
-    private void validateEnd(AtlasObjectId end) throws AtlasBaseException {
-        String              guid             = end.getGuid();
-        String              typeName         = end.getTypeName();
-        Map<String, Object> uniqueAttributes = end.getUniqueAttributes();
-        AtlasVertex         endVertex        = AtlasGraphUtilsV1.findByGuid(guid);
-
-        if (!AtlasTypeUtil.isValidGuid(guid) || endVertex == null) {
-            throw new AtlasBaseException(AtlasErrorCode.INSTANCE_GUID_NOT_FOUND, guid);
-        } else if (MapUtils.isNotEmpty(uniqueAttributes))  {
-            AtlasEntityType entityType = typeRegistry.getEntityTypeByName(typeName);
-
-            if (AtlasGraphUtilsV1.findByUniqueAttributes(entityType, uniqueAttributes) == null) {
-                throw new AtlasBaseException(AtlasErrorCode.INSTANCE_BY_UNIQUE_ATTRIBUTE_NOT_FOUND, typeName, uniqueAttributes.toString());
+    /**
+     * Validate the ends of the passed relationship
+     * @param relationship
+     * @throws AtlasBaseException
+     */
+    private void validateEnds(AtlasRelationship relationship) throws AtlasBaseException {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("validateEnds entry relationship:" + relationship);
+        }
+        List<AtlasObjectId> ends = new ArrayList<>();
+        List<AtlasRelationshipEndDef> endDefs = new ArrayList<>();
+        String relationshipTypeName = relationship.getTypeName();
+        AtlasRelationshipDef relationshipDef = typeRegistry.getRelationshipDefByName(relationshipTypeName);
+        ends.add(relationship.getEnd1());
+        ends.add(relationship.getEnd2());
+        endDefs.add(relationshipDef.getEndDef1());
+        endDefs.add(relationshipDef.getEndDef2());
+
+        for (int i = 0; i < ends.size(); i++) {
+            AtlasObjectId end = ends.get(i);
+            String guid = end.getGuid();
+            String typeName = end.getTypeName();
+            Map<String, Object> uniqueAttributes = end.getUniqueAttributes();
+            AtlasVertex endVertex = AtlasGraphUtilsV1.findByGuid(guid);
+
+            if (!AtlasTypeUtil.isValidGuid(guid) || endVertex == null) {
+                throw new AtlasBaseException(AtlasErrorCode.INSTANCE_GUID_NOT_FOUND, guid);
+            } else if (MapUtils.isNotEmpty(uniqueAttributes)) {
+                AtlasEntityType entityType = typeRegistry.getEntityTypeByName(typeName);
+                if (AtlasGraphUtilsV1.findByUniqueAttributes(entityType, uniqueAttributes) == null) {
+                    throw new AtlasBaseException(AtlasErrorCode.INSTANCE_BY_UNIQUE_ATTRIBUTE_NOT_FOUND, typeName, uniqueAttributes.toString());
+                }
+            } else {
+                // check whether the guid is the correct type
+                String vertexTypeName = endVertex.getProperty(Constants.TYPE_NAME_PROPERTY_KEY, String.class);
+                if (!Objects.equals(vertexTypeName, typeName)) {
+                    String attrName = endDefs.get(i).getName();
+                    throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIP_INVALID_ENDTYPE, attrName, guid, vertexTypeName, typeName);
+                }
             }
         }
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("validateEnds exit successfully validated relationship:" + relationship);
+        }
     }
 
     private void validateAndNormalize(AtlasRelationship relationship) throws AtlasBaseException {
@@ -341,6 +365,9 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
     }
 
     private String getRelationshipEdgeLabel(AtlasVertex fromVertex, AtlasVertex toVertex, AtlasRelationship relationship) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("validateEnds entry relationship:"+relationship);
+        }
         AtlasRelationshipType   relationshipType   = typeRegistry.getRelationshipTypeByName(relationship.getTypeName());
         String                  ret                = relationshipType.getRelationshipDef().getRelationshipLabel();
         AtlasRelationshipEndDef endDef1            = relationshipType.getRelationshipDef().getEndDef1();