falcon-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From venkat...@apache.org
Subject [2/3] git commit: FALCON-366 Add a REST API to get properties for a given vertex. Contributed by Venkatesh Seetharam
Date Fri, 21 Mar 2014 22:24:04 GMT
FALCON-366 Add a REST API to get properties for a given vertex. Contributed by Venkatesh Seetharam


Project: http://git-wip-us.apache.org/repos/asf/incubator-falcon/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-falcon/commit/4e8065c5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-falcon/tree/4e8065c5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-falcon/diff/4e8065c5

Branch: refs/heads/master
Commit: 4e8065c58ae94ebeb2905e29b8462df468be15cd
Parents: 8f7c0dc
Author: Venkatesh Seetharam <venkatesh@hortonworks.com>
Authored: Fri Mar 21 14:10:50 2014 -0700
Committer: Venkatesh Seetharam <venkatesh@hortonworks.com>
Committed: Fri Mar 21 14:10:50 2014 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |   2 +
 .../EntityRelationshipGraphBuilder.java         |  69 +-
 .../InstanceRelationshipGraphBuilder.java       |  47 +-
 .../falcon/metadata/MetadataMappingService.java |   8 +-
 .../metadata/RelationshipGraphBuilder.java      |  73 +-
 .../falcon/metadata/RelationshipLabel.java      |  50 ++
 .../falcon/metadata/RelationshipProperty.java   |  64 ++
 .../falcon/metadata/RelationshipType.java       |  63 ++
 common/src/main/resources/startup.properties    |   2 +-
 .../metadata/MetadataMappingServiceTest.java    | 158 +++--
 .../metadata/LineageMetadataResource.java       | 136 +++-
 .../metadata/LineageMetadataResourceTest.java   | 663 +++++++++++++++++++
 12 files changed, 1124 insertions(+), 211 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 91d2ad9..622d767 100755
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -78,6 +78,8 @@ Trunk (Unreleased)
     FALCON-342 Add ability to set maxMaps on replication (Michael Miklavcic
     via Venkatesh Seetharam)
 
+    FALCON-366 Add a REST API to get properties for a given vertex (Venkatesh Seetharam)
+
   OPTIMIZATIONS
     FALCON-123 Improve build speeds in falcon. (Srikanth Sundarrajan via Shwetha GS)
 

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
index 4e94f61..2d3ec95 100644
--- a/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/EntityRelationshipGraphBuilder.java
@@ -41,11 +41,6 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
 
     private static final Logger LOG = Logger.getLogger(EntityRelationshipGraphBuilder.class);
 
-    // entity vertex types
-    public static final String CLUSTER_ENTITY_TYPE = "cluster-entity";
-    public static final String FEED_ENTITY_TYPE = "feed-entity";
-    public static final String PROCESS_ENTITY_TYPE = "process-entity";
-
 
     public EntityRelationshipGraphBuilder(Graph graph, boolean preserveHistory) {
         super(graph, preserveHistory);
@@ -53,29 +48,28 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
 
     public void addClusterEntity(Cluster clusterEntity) {
         LOG.info("Adding cluster entity: " + clusterEntity.getName());
-        Vertex clusterVertex = addVertex(clusterEntity.getName(), CLUSTER_ENTITY_TYPE);
+        Vertex clusterVertex = addVertex(clusterEntity.getName(), RelationshipType.CLUSTER_ENTITY);
 
-        // addUserRelation(clusterVertex, "created-by");  // audit info
         addColoRelation(clusterEntity.getColo(), clusterVertex);
         addDataClassification(clusterEntity.getTags(), clusterVertex);
     }
 
     public void addFeedEntity(Feed feed) {
         LOG.info("Adding feed entity: " + feed.getName());
-        Vertex feedVertex = addVertex(feed.getName(), FEED_ENTITY_TYPE);
+        Vertex feedVertex = addVertex(feed.getName(), RelationshipType.FEED_ENTITY);
 
         addUserRelation(feedVertex);
         addDataClassification(feed.getTags(), feedVertex);
         addGroups(feed.getGroups(), feedVertex);
 
         for (org.apache.falcon.entity.v0.feed.Cluster feedCluster : feed.getClusters().getClusters()) {
-            addRelationToCluster(feedVertex, feedCluster.getName(), FEED_CLUSTER_EDGE_LABEL);
+            addRelationToCluster(feedVertex, feedCluster.getName(), RelationshipLabel.FEED_CLUSTER_EDGE);
         }
     }
 
     public void updateFeedEntity(Feed oldFeed, Feed newFeed) {
         LOG.info("Updating feed entity: " + newFeed.getName());
-        Vertex feedEntityVertex = findVertex(oldFeed.getName(), FEED_ENTITY_TYPE);
+        Vertex feedEntityVertex = findVertex(oldFeed.getName(), RelationshipType.FEED_ENTITY);
         if (feedEntityVertex == null) {
             // todo - throw new IllegalStateException(oldFeed.getName() + " entity vertex must exist.");
             LOG.error("Illegal State: Feed entity vertex must exist for " + oldFeed.getName());
@@ -91,24 +85,23 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
     public void addProcessEntity(Process process) {
         String processName = process.getName();
         LOG.info("Adding process entity: " + processName);
-        Vertex processVertex = addVertex(processName, PROCESS_ENTITY_TYPE);
+        Vertex processVertex = addVertex(processName, RelationshipType.PROCESS_ENTITY);
         addWorkflowProperties(process.getWorkflow(), processVertex, processName);
 
         addUserRelation(processVertex);
         addDataClassification(process.getTags(), processVertex);
 
         for (org.apache.falcon.entity.v0.process.Cluster cluster : process.getClusters().getClusters()) {
-            addRelationToCluster(processVertex, cluster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+            addRelationToCluster(processVertex, cluster.getName(), RelationshipLabel.PROCESS_CLUSTER_EDGE);
         }
 
         addInputFeeds(process.getInputs(), processVertex);
         addOutputFeeds(process.getOutputs(), processVertex);
-        // addWorkflow(process.getWorkflow(), processVertex, processName);
     }
 
     public void updateProcessEntity(Process oldProcess, Process newProcess) {
         LOG.info("Updating process entity: " + newProcess.getName());
-        Vertex processEntityVertex = findVertex(oldProcess.getName(), PROCESS_ENTITY_TYPE);
+        Vertex processEntityVertex = findVertex(oldProcess.getName(), RelationshipType.PROCESS_ENTITY);
         if (processEntityVertex == null) {
             // todo - throw new IllegalStateException(oldProcess.getName() + " entity vertex must exist");
             LOG.error("Illegal State: Process entity vertex must exist for " + oldProcess.getName());
@@ -125,19 +118,19 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
     }
 
     public void addColoRelation(String colo, Vertex fromVertex) {
-        Vertex coloVertex = addVertex(colo, COLO_TYPE);
-        addEdge(fromVertex, coloVertex, CLUSTER_COLO_LABEL);
+        Vertex coloVertex = addVertex(colo, RelationshipType.COLO);
+        addEdge(fromVertex, coloVertex, RelationshipLabel.CLUSTER_COLO.getName());
     }
 
-    public void addRelationToCluster(Vertex fromVertex, String clusterName, String edgeLabel) {
-        Vertex clusterVertex = findVertex(clusterName, CLUSTER_ENTITY_TYPE);
+    public void addRelationToCluster(Vertex fromVertex, String clusterName, RelationshipLabel edgeLabel) {
+        Vertex clusterVertex = findVertex(clusterName, RelationshipType.CLUSTER_ENTITY);
         if (clusterVertex == null) { // cluster must exist before adding other entities
             // todo - throw new IllegalStateException("Cluster entity vertex must exist: " + clusterName);
             LOG.error("Illegal State: Cluster entity vertex must exist for " + clusterName);
             return;
         }
 
-        addEdge(fromVertex, clusterVertex, edgeLabel);
+        addEdge(fromVertex, clusterVertex, edgeLabel.getName());
     }
 
     public void addInputFeeds(Inputs inputs, Vertex processVertex) {
@@ -146,7 +139,7 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
         }
 
         for (Input input : inputs.getInputs()) {
-            addProcessFeedEdge(processVertex, input.getFeed(), FEED_PROCESS_EDGE_LABEL);
+            addProcessFeedEdge(processVertex, input.getFeed(), RelationshipLabel.FEED_PROCESS_EDGE);
         }
     }
 
@@ -156,12 +149,12 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
         }
 
         for (Output output : outputs.getOutputs()) {
-            addProcessFeedEdge(processVertex, output.getFeed(), PROCESS_FEED_EDGE_LABEL);
+            addProcessFeedEdge(processVertex, output.getFeed(), RelationshipLabel.PROCESS_FEED_EDGE);
         }
     }
 
-    public void addProcessFeedEdge(Vertex processVertex, String feedName, String edgeLabel) {
-        Vertex feedVertex = findVertex(feedName, FEED_ENTITY_TYPE);
+    public void addProcessFeedEdge(Vertex processVertex, String feedName, RelationshipLabel edgeLabel) {
+        Vertex feedVertex = findVertex(feedName, RelationshipType.FEED_ENTITY);
         if (feedVertex == null) {
             // todo - throw new IllegalStateException("Feed entity vertex must exist: " + feedName);
             LOG.error("Illegal State: Feed entity vertex must exist for " + feedName);
@@ -174,7 +167,7 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
     public void addWorkflowProperties(Workflow workflow, Vertex processVertex, String processName) {
         processVertex.setProperty(LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
                 ProcessHelper.getProcessWorkflowName(workflow.getName(), processName));
-        processVertex.setProperty(VERSION_PROPERTY_KEY, workflow.getVersion());
+        processVertex.setProperty(RelationshipProperty.VERSION.getName(), workflow.getVersion());
         processVertex.setProperty(LineageArgs.USER_WORKFLOW_ENGINE.getOptionName(), workflow.getEngine().value());
     }
 
@@ -229,7 +222,7 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
 
         String[] oldGroupTags = groups.split(",");
         for (String groupTag : oldGroupTags) {
-            removeEdge(entityVertex, groupTag, RelationshipGraphBuilder.GROUPS_LABEL);
+            removeEdge(entityVertex, groupTag, RelationshipLabel.GROUPS.getName());
         }
     }
 
@@ -247,12 +240,14 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
 
         // remove edges to old clusters
         for (org.apache.falcon.entity.v0.feed.Cluster oldCuster : oldClusters) {
-            removeEdge(feedEntityVertex, oldCuster.getName(), FEED_CLUSTER_EDGE_LABEL);
+            removeEdge(feedEntityVertex, oldCuster.getName(),
+                    RelationshipLabel.FEED_CLUSTER_EDGE.getName());
         }
 
         // add edges to new clusters
         for (org.apache.falcon.entity.v0.feed.Cluster newCluster : newClusters) {
-            addRelationToCluster(feedEntityVertex, newCluster.getName(), FEED_CLUSTER_EDGE_LABEL);
+            addRelationToCluster(feedEntityVertex, newCluster.getName(),
+                    RelationshipLabel.FEED_CLUSTER_EDGE);
         }
     }
 
@@ -288,12 +283,14 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
 
         // remove old clusters
         for (org.apache.falcon.entity.v0.process.Cluster oldCuster : oldClusters) {
-            removeEdge(processEntityVertex, oldCuster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+            removeEdge(processEntityVertex, oldCuster.getName(),
+                    RelationshipLabel.PROCESS_CLUSTER_EDGE.getName());
         }
 
         // add new clusters
         for (org.apache.falcon.entity.v0.process.Cluster newCluster : newClusters) {
-            addRelationToCluster(processEntityVertex, newCluster.getName(), PROCESS_CLUSTER_EDGE_LABEL);
+            addRelationToCluster(processEntityVertex, newCluster.getName(),
+                    RelationshipLabel.PROCESS_CLUSTER_EDGE);
         }
     }
 
@@ -357,7 +354,7 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
         }
 
         for (Input input : inputs.getInputs()) {
-            removeProcessFeedEdge(processVertex, input.getFeed(), FEED_PROCESS_EDGE_LABEL);
+            removeProcessFeedEdge(processVertex, input.getFeed(), RelationshipLabel.FEED_PROCESS_EDGE);
         }
     }
 
@@ -367,22 +364,22 @@ public class EntityRelationshipGraphBuilder extends RelationshipGraphBuilder {
         }
 
         for (Output output : outputs.getOutputs()) {
-            removeProcessFeedEdge(processVertex, output.getFeed(), PROCESS_FEED_EDGE_LABEL);
+            removeProcessFeedEdge(processVertex, output.getFeed(), RelationshipLabel.PROCESS_FEED_EDGE);
         }
     }
 
-    public void removeProcessFeedEdge(Vertex processVertex, String feedName, String edgeLabel) {
-        Vertex feedVertex = findVertex(feedName, FEED_ENTITY_TYPE);
+    public void removeProcessFeedEdge(Vertex processVertex, String feedName, RelationshipLabel edgeLabel) {
+        Vertex feedVertex = findVertex(feedName, RelationshipType.FEED_ENTITY);
         if (feedVertex == null) {
             // todo - throw new IllegalStateException("Feed entity vertex must exist: " + feedName);
             LOG.error("Illegal State: Feed entity vertex must exist for " + feedName);
             return;
         }
 
-        if (edgeLabel.equals(FEED_PROCESS_EDGE_LABEL)) {
-            removeEdge(feedVertex, processVertex, edgeLabel);
+        if (edgeLabel == RelationshipLabel.FEED_PROCESS_EDGE) {
+            removeEdge(feedVertex, processVertex, edgeLabel.getName());
         } else {
-            removeEdge(processVertex, feedVertex, edgeLabel);
+            removeEdge(processVertex, feedVertex, edgeLabel.getName());
         }
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
index 7efa74a..eb591c0 100644
--- a/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/InstanceRelationshipGraphBuilder.java
@@ -47,10 +47,6 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
     private static final String PROCESS_INSTANCE_FORMAT = "yyyy-MM-dd-HH-mm"; // nominal time
     private static final String FEED_INSTANCE_FORMAT = "yyyyMMddHHmm"; // computed
 
-    // instance vertex types
-    public static final String FEED_INSTANCE_TYPE = "feed-instance";
-    public static final String PROCESS_INSTANCE_TYPE = "process-instance";
-
     // process workflow properties from message
     private static final String[] INSTANCE_WORKFLOW_PROPERTIES = {
         LineageArgs.USER_WORKFLOW_NAME.getOptionName(),
@@ -62,8 +58,6 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
         LineageArgs.USER_SUBFLOW_ID.getOptionName(),
     };
 
-    // instance edge labels
-    public static final String INSTANCE_ENTITY_EDGE_LABEL = "instance-of";
 
     public InstanceRelationshipGraphBuilder(Graph graph, boolean preserveHistory) {
         super(graph, preserveHistory);
@@ -75,16 +69,16 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
                 lineageMetadata.get(LineageArgs.NOMINAL_TIME.getOptionName()));
         LOG.info("Adding process instance: " + processInstanceName);
 
-        String timestamp = lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName());
-        Vertex processInstance = addVertex(processInstanceName, PROCESS_INSTANCE_TYPE, timestamp);
+        String timestamp = getTimestamp(lineageMetadata);
+        Vertex processInstance = addVertex(processInstanceName, RelationshipType.PROCESS_INSTANCE, timestamp);
         addWorkflowInstanceProperties(processInstance, lineageMetadata);
 
         addInstanceToEntity(processInstance, entityName,
-                EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE, INSTANCE_ENTITY_EDGE_LABEL);
+                RelationshipType.PROCESS_ENTITY, RelationshipLabel.INSTANCE_ENTITY_EDGE);
         addInstanceToEntity(processInstance, lineageMetadata.get(LineageArgs.CLUSTER.getOptionName()),
-                EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE, PROCESS_CLUSTER_EDGE_LABEL);
-        addInstanceToEntity(processInstance,
-                lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()), USER_TYPE, USER_LABEL);
+                RelationshipType.CLUSTER_ENTITY, RelationshipLabel.PROCESS_CLUSTER_EDGE);
+        addInstanceToEntity(processInstance, lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()),
+                RelationshipType.USER, RelationshipLabel.USER);
 
         if (isPreserveHistory()) {
             Process process = ConfigurationStore.get().get(EntityType.PROCESS, entityName);
@@ -94,12 +88,17 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
         return processInstance;
     }
 
+    private String getTimestamp(Map<String, String> lineageMetadata) {
+        String timestamp = lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName());
+        return SchemaHelper.formatDateUTCToISO8601(timestamp, PROCESS_INSTANCE_FORMAT);
+    }
+
     public void addWorkflowInstanceProperties(Vertex processInstance,
                                               Map<String, String> lineageMetadata) {
         for (String instanceWorkflowProperty : INSTANCE_WORKFLOW_PROPERTIES) {
             addProperty(processInstance, lineageMetadata, instanceWorkflowProperty);
         }
-        processInstance.setProperty(VERSION_PROPERTY_KEY,
+        processInstance.setProperty(RelationshipProperty.VERSION.getName(),
                 lineageMetadata.get(LineageArgs.USER_WORKFLOW_VERSION.getOptionName()));
     }
 
@@ -110,7 +109,7 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
     }
 
     public void addInstanceToEntity(Vertex instanceVertex, String entityName,
-                                    String entityType, String edgeLabel) {
+                                    RelationshipType entityType, RelationshipLabel edgeLabel) {
         Vertex entityVertex = findVertex(entityName, entityType);
         LOG.info("Vertex exists? name=" + entityName + ", type=" + entityType + ", v=" + entityVertex);
         if (entityVertex == null) {
@@ -119,7 +118,7 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
             return;
         }
 
-        addEdge(instanceVertex, entityVertex, edgeLabel);
+        addEdge(instanceVertex, entityVertex, edgeLabel.getName());
     }
 
     public void addOutputFeedInstances(Map<String, String> lineageMetadata,
@@ -134,7 +133,7 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
                 lineageMetadata.get(LineageArgs.FEED_INSTANCE_PATHS.getOptionName()).split(",");
 
         addFeedInstances(outputFeedNames, outputFeedInstancePaths,
-                processInstance, PROCESS_FEED_EDGE_LABEL, lineageMetadata);
+                processInstance, RelationshipLabel.PROCESS_FEED_EDGE, lineageMetadata);
     }
 
     public void addInputFeedInstances(Map<String, String> lineageMetadata,
@@ -150,11 +149,11 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
                 lineageMetadata.get(LineageArgs.INPUT_FEED_PATHS.getOptionName()).split("#");
 
         addFeedInstances(inputFeedNames, inputFeedInstancePaths,
-                processInstance, FEED_PROCESS_EDGE_LABEL, lineageMetadata);
+                processInstance, RelationshipLabel.FEED_PROCESS_EDGE, lineageMetadata);
     }
 
     public void addFeedInstances(String[] feedNames, String[] feedInstancePaths,
-                                  Vertex processInstance, String edgeLabel,
+                                  Vertex processInstance, RelationshipLabel edgeLabel,
                                   Map<String, String> lineageMetadata) throws FalconException {
         String clusterName = lineageMetadata.get(LineageArgs.CLUSTER.getOptionName());
 
@@ -166,17 +165,17 @@ public class InstanceRelationshipGraphBuilder extends RelationshipGraphBuilder {
                     + feedInstancePath + ", in cluster: " + clusterName);
             String feedInstanceName = getFeedInstanceName(feedName, clusterName, feedInstancePath);
             LOG.info("Adding feed instance: " + feedInstanceName);
-            Vertex feedInstance = addVertex(feedInstanceName, FEED_INSTANCE_TYPE,
-                    lineageMetadata.get(LineageArgs.TIMESTAMP.getOptionName()));
+            Vertex feedInstance = addVertex(feedInstanceName, RelationshipType.FEED_INSTANCE,
+                    getTimestamp(lineageMetadata));
 
             addProcessFeedEdge(processInstance, feedInstance, edgeLabel);
 
             addInstanceToEntity(feedInstance, feedName,
-                    EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE, INSTANCE_ENTITY_EDGE_LABEL);
+                    RelationshipType.FEED_ENTITY, RelationshipLabel.INSTANCE_ENTITY_EDGE);
             addInstanceToEntity(feedInstance, lineageMetadata.get(LineageArgs.CLUSTER.getOptionName()),
-                    EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE, FEED_CLUSTER_EDGE_LABEL);
-            addInstanceToEntity(feedInstance,
-                    lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()), USER_TYPE, USER_LABEL);
+                    RelationshipType.CLUSTER_ENTITY, RelationshipLabel.FEED_CLUSTER_EDGE);
+            addInstanceToEntity(feedInstance, lineageMetadata.get(LineageArgs.WORKFLOW_USER.getOptionName()),
+                    RelationshipType.USER, RelationshipLabel.USER);
 
             if (isPreserveHistory()) {
                 Feed feed = ConfigurationStore.get().get(EntityType.FEED, feedName);

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
index 3a6db91..21c22de 100644
--- a/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
+++ b/common/src/main/java/org/apache/falcon/metadata/MetadataMappingService.java
@@ -129,13 +129,13 @@ public class MetadataMappingService implements FalconService, ConfigurationChang
         LOG.info("Indexes does not exist, Creating indexes for graph");
         // todo - externalize this
         makeNameKeyIndex();
-        makeKeyIndex(RelationshipGraphBuilder.TYPE_PROPERTY_KEY);
-        makeKeyIndex(RelationshipGraphBuilder.TIMESTAMP_PROPERTY_KEY);
-        makeKeyIndex(RelationshipGraphBuilder.VERSION_PROPERTY_KEY);
+        makeKeyIndex(RelationshipProperty.TYPE.getName());
+        makeKeyIndex(RelationshipProperty.TIMESTAMP.getName());
+        makeKeyIndex(RelationshipProperty.VERSION.getName());
     }
 
     private void makeNameKeyIndex() {
-        getTitanGraph().makeKey(RelationshipGraphBuilder.NAME_PROPERTY_KEY)
+        getTitanGraph().makeKey(RelationshipProperty.NAME.getName())
                 .dataType(String.class)
                 .indexed(Vertex.class)
                 .indexed(Edge.class)

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
index 5c0f8cb..7baeeec 100644
--- a/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipGraphBuilder.java
@@ -23,14 +23,13 @@ import com.tinkerpop.blueprints.Edge;
 import com.tinkerpop.blueprints.Graph;
 import com.tinkerpop.blueprints.GraphQuery;
 import com.tinkerpop.blueprints.Vertex;
+import org.apache.falcon.entity.v0.SchemaHelper;
 import org.apache.falcon.security.CurrentUser;
 import org.apache.log4j.Logger;
 
-import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.Iterator;
 import java.util.Map;
-import java.util.TimeZone;
 
 /**
  * Base class for Metadata relationship mapping helper.
@@ -39,29 +38,6 @@ public abstract class RelationshipGraphBuilder {
 
     private static final Logger LOG = Logger.getLogger(RelationshipGraphBuilder.class);
 
-    // vertex property keys
-    public static final String NAME_PROPERTY_KEY = "name";
-    public static final String TYPE_PROPERTY_KEY = "type";
-    public static final String TIMESTAMP_PROPERTY_KEY = "timestamp";
-    public static final String VERSION_PROPERTY_KEY = "version";
-
-    // vertex types
-    public static final String USER_TYPE = "user";
-    public static final String COLO_TYPE = "data-center";
-    public static final String TAGS_TYPE = "classification";
-    public static final String GROUPS_TYPE = "group";
-
-    // edge labels
-    public static final String USER_LABEL = "owned-by";
-    public static final String CLUSTER_COLO_LABEL = "collocated";
-    public static final String GROUPS_LABEL = "grouped-as";
-
-    // entity edge labels
-    public static final String FEED_CLUSTER_EDGE_LABEL = "stored-in";
-    public static final String PROCESS_CLUSTER_EDGE_LABEL = "runs-on";
-    public static final String FEED_PROCESS_EDGE_LABEL = "input";
-    public static final String PROCESS_FEED_EDGE_LABEL = "output";
-
     /**
      * A blueprints graph.
      */
@@ -86,7 +62,7 @@ public abstract class RelationshipGraphBuilder {
         return preserveHistory;
     }
 
-    public Vertex addVertex(String name, String type) {
+    public Vertex addVertex(String name, RelationshipType type) {
         Vertex vertex = findVertex(name, type);
         if (vertex != null) {
             if (LOG.isDebugEnabled()) {
@@ -99,7 +75,7 @@ public abstract class RelationshipGraphBuilder {
         return createVertex(name, type);
     }
 
-    protected Vertex addVertex(String name, String type, String timestamp) {
+    protected Vertex addVertex(String name, RelationshipType type, String timestamp) {
         Vertex vertex = findVertex(name, type);
         if (vertex != null) {
             if (LOG.isDebugEnabled()) {
@@ -112,31 +88,31 @@ public abstract class RelationshipGraphBuilder {
         return createVertex(name, type, timestamp);
     }
 
-    protected Vertex findVertex(String name, String type) {
+    protected Vertex findVertex(String name, RelationshipType type) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Finding vertex for: name=" + name + ", type=" + type);
         }
 
         GraphQuery query = graph.query()
-                .has(NAME_PROPERTY_KEY, name)
-                .has(TYPE_PROPERTY_KEY, type);
+                .has(RelationshipProperty.NAME.getName(), name)
+                .has(RelationshipProperty.TYPE.getName(), type.getName());
         Iterator<Vertex> results = query.vertices().iterator();
         return results.hasNext() ? results.next() : null;  // returning one since name is unique
     }
 
-    protected Vertex createVertex(String name, String type) {
+    protected Vertex createVertex(String name, RelationshipType type) {
         return createVertex(name, type, getCurrentTimeStamp());
     }
 
-    protected Vertex createVertex(String name, String type, String timestamp) {
+    protected Vertex createVertex(String name, RelationshipType type, String timestamp) {
         if (LOG.isDebugEnabled()) {
             LOG.debug("Creating a new vertex for: name=" + name + ", type=" + type);
         }
 
         Vertex vertex = graph.addVertex(null);
-        vertex.setProperty(NAME_PROPERTY_KEY, name);
-        vertex.setProperty(TYPE_PROPERTY_KEY, type);
-        vertex.setProperty(TIMESTAMP_PROPERTY_KEY, timestamp);
+        vertex.setProperty(RelationshipProperty.NAME.getName(), name);
+        vertex.setProperty(RelationshipProperty.TYPE.getName(), type.getName());
+        vertex.setProperty(RelationshipProperty.TIMESTAMP.getName(), timestamp);
 
         return vertex;
     }
@@ -161,13 +137,13 @@ public abstract class RelationshipGraphBuilder {
     }
 
     protected Edge findEdge(Vertex fromVertex, Vertex toVertex, String edgeLabel) {
-        return findEdge(fromVertex, toVertex.getProperty(NAME_PROPERTY_KEY), edgeLabel);
+        return findEdge(fromVertex, toVertex.getProperty(RelationshipProperty.NAME.getName()), edgeLabel);
     }
 
     protected Edge findEdge(Vertex fromVertex, Object toVertexName, String edgeLabel) {
         Edge edgeToFind = null;
         for (Edge edge : fromVertex.getEdges(Direction.OUT, edgeLabel)) {
-            if (edge.getVertex(Direction.IN).getProperty(NAME_PROPERTY_KEY).equals(toVertexName)) {
+            if (edge.getVertex(Direction.IN).getProperty(RelationshipProperty.NAME.getName()).equals(toVertexName)) {
                 edgeToFind = edge;
                 break;
             }
@@ -177,11 +153,11 @@ public abstract class RelationshipGraphBuilder {
     }
 
     protected void addUserRelation(Vertex fromVertex) {
-        addUserRelation(fromVertex, USER_LABEL);
+        addUserRelation(fromVertex, RelationshipLabel.USER.getName());
     }
 
     protected void addUserRelation(Vertex fromVertex, String edgeLabel) {
-        Vertex relationToUserVertex = addVertex(CurrentUser.getUser(), USER_TYPE);
+        Vertex relationToUserVertex = addVertex(CurrentUser.getUser(), RelationshipType.USER);
         addEdge(fromVertex, relationToUserVertex, edgeLabel);
     }
 
@@ -196,7 +172,7 @@ public abstract class RelationshipGraphBuilder {
             String tagKey = tag.substring(0, index);
             String tagValue = tag.substring(index + 1, tag.length());
 
-            Vertex tagValueVertex = addVertex(tagValue, TAGS_TYPE);
+            Vertex tagValueVertex = addVertex(tagValue, RelationshipType.TAGS);
             addEdge(entityVertex, tagValueVertex, tagKey);
         }
     }
@@ -208,23 +184,22 @@ public abstract class RelationshipGraphBuilder {
 
         String[] groupTags = groups.split(",");
         for (String groupTag : groupTags) {
-            Vertex groupVertex = addVertex(groupTag, GROUPS_TYPE);
-            addEdge(fromVertex, groupVertex, GROUPS_LABEL);
+            Vertex groupVertex = addVertex(groupTag, RelationshipType.GROUPS);
+            addEdge(fromVertex, groupVertex, RelationshipLabel.GROUPS.getName());
         }
     }
 
-    protected void addProcessFeedEdge(Vertex processVertex, Vertex feedVertex, String edgeLabel) {
-        if (edgeLabel.equals(FEED_PROCESS_EDGE_LABEL)) {
-            addEdge(feedVertex, processVertex, edgeLabel);
+    protected void addProcessFeedEdge(Vertex processVertex, Vertex feedVertex,
+                                      RelationshipLabel edgeLabel) {
+        if (edgeLabel == RelationshipLabel.FEED_PROCESS_EDGE) {
+            addEdge(feedVertex, processVertex, edgeLabel.getName());
         } else {
-            addEdge(processVertex, feedVertex, edgeLabel);
+            addEdge(processVertex, feedVertex, edgeLabel.getName());
         }
     }
 
     protected String getCurrentTimeStamp() {
-        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-HH-mm");
-        dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
-        return dateFormat.format(new Date());
+        return SchemaHelper.formatDateUTC(new Date());
     }
 
     protected void addProperty(Vertex vertex, Map<String, String> lineageMetadata, String optionName) {

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/RelationshipLabel.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipLabel.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipLabel.java
new file mode 100644
index 0000000..2e8b864
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipLabel.java
@@ -0,0 +1,50 @@
+/**
+ * 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.falcon.metadata;
+
+/**
+ * Enumerates Relationship edge labels.
+ */
+public enum RelationshipLabel {
+
+    // entity edge labels
+    FEED_CLUSTER_EDGE("stored-in"),
+    PROCESS_CLUSTER_EDGE("runs-on"),
+    FEED_PROCESS_EDGE("input"),
+    PROCESS_FEED_EDGE("output"),
+
+    // instance edge labels
+    INSTANCE_ENTITY_EDGE("instance-of"),
+
+    // edge labels
+    CLUSTER_COLO("collocated"),
+    USER("owned-by"),
+    GROUPS("grouped-as");
+
+
+    private final String name;
+
+    RelationshipLabel(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/RelationshipProperty.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipProperty.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipProperty.java
new file mode 100644
index 0000000..ff437d9
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipProperty.java
@@ -0,0 +1,64 @@
+/**
+ * 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.falcon.metadata;
+
+/**
+ * Enumerates Relationship property keys.
+ */
+public enum RelationshipProperty {
+
+    // vertex property keys - indexed
+    NAME("name"),
+    TYPE("type"),
+    TIMESTAMP("timestamp"),
+    VERSION("version"),
+
+    // workflow properties
+    USER_WORKFLOW_ENGINE("userWorkflowEngine", "user workflow engine type"),
+    USER_WORKFLOW_NAME("userWorkflowName", "user workflow name"),
+    USER_WORKFLOW_VERSION("userWorkflowVersion", "user workflow version"),
+
+    // workflow instance properties
+    WORKFLOW_ID("workflowId", "current workflow-id of the instance"),
+    RUN_ID("runId", "current run-id of the instance"),
+    STATUS("status", "status of the user workflow instance"),
+    WF_ENGINE_URL("workflowEngineUrl", "url of workflow engine server, ex: oozie"),
+    USER_SUBFLOW_ID("subflowId", "external id of user workflow");
+
+
+    private final String name;
+    private final String description;
+
+    RelationshipProperty(String name) {
+        this(name, name);
+    }
+
+    RelationshipProperty(String name, String description) {
+        this.name = name;
+        this.description = description;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/java/org/apache/falcon/metadata/RelationshipType.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/falcon/metadata/RelationshipType.java b/common/src/main/java/org/apache/falcon/metadata/RelationshipType.java
new file mode 100644
index 0000000..075715f
--- /dev/null
+++ b/common/src/main/java/org/apache/falcon/metadata/RelationshipType.java
@@ -0,0 +1,63 @@
+/**
+ * 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.falcon.metadata;
+
+/**
+ * Enumerates Relationship types.
+ */
+public enum RelationshipType {
+
+    // entity vertex types
+    CLUSTER_ENTITY("cluster-entity"),
+    FEED_ENTITY("feed-entity"),
+    PROCESS_ENTITY("process-entity"),
+
+    // instance vertex types
+    FEED_INSTANCE("feed-instance"),
+    PROCESS_INSTANCE("process-instance"),
+
+    // Misc vertex types
+    USER("user"),
+    COLO("data-center"),
+    TAGS("classification"),
+    GROUPS("group");
+
+
+    private final String name;
+
+    RelationshipType(java.lang.String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public static RelationshipType fromString(String value) {
+        if (value != null) {
+            for (RelationshipType type : RelationshipType.values()) {
+                if (value.equals(type.getName())) {
+                    return type;
+                }
+            }
+        }
+
+        throw new IllegalArgumentException("No constant with value " + value + " found");
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
index 457b3a6..ccd969f 100644
--- a/common/src/main/resources/startup.properties
+++ b/common/src/main/resources/startup.properties
@@ -83,7 +83,7 @@ debug.libext.process.paths=${falcon.libext}
 # Graph Storage
 *.falcon.graph.storage.directory=${user.dir}/target/graphdb
 *.falcon.graph.storage.backend=berkeleyje
-*.falcon.graph.serialize.path=/${falcon.home}/data
+*.falcon.graph.serialize.path=${user.dir}/target/graphdb
 *.falcon.graph.preserve.history=false
 
 

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
----------------------------------------------------------------------
diff --git a/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
index dab308d..f016f25 100644
--- a/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
+++ b/common/src/test/java/org/apache/falcon/metadata/MetadataMappingServiceTest.java
@@ -138,7 +138,7 @@ public class MetadataMappingServiceTest {
         clusterEntity = buildCluster(CLUSTER_ENTITY_NAME, COLO_NAME, "classification=production");
         configStore.publish(EntityType.CLUSTER, clusterEntity);
 
-        verifyEntityWasAddedToGraph(CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(CLUSTER_ENTITY_NAME, RelationshipType.CLUSTER_ENTITY);
         verifyClusterEntityEdges();
 
         Assert.assertEquals(getVerticesCount(service.getGraph()), 3); // +3 = cluster, colo, tag
@@ -151,7 +151,7 @@ public class MetadataMappingServiceTest {
                 Storage.TYPE.FILESYSTEM, "/falcon/impression-feed/${YEAR}${MONTH}${DAY}");
         configStore.publish(EntityType.FEED, impressionsFeed);
         inputFeeds.add(impressionsFeed);
-        verifyEntityWasAddedToGraph(impressionsFeed.getName(), EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(impressionsFeed.getName(), RelationshipType.FEED_ENTITY);
         verifyFeedEntityEdges(impressionsFeed.getName());
         Assert.assertEquals(getVerticesCount(service.getGraph()), 7); // +4 = feed, tag, group, user
         Assert.assertEquals(getEdgesCount(service.getGraph()), 6); // +4 = cluster, tag, group, user
@@ -160,8 +160,7 @@ public class MetadataMappingServiceTest {
                 "analytics", Storage.TYPE.FILESYSTEM, "/falcon/clicks-feed/${YEAR}${MONTH}${DAY}");
         configStore.publish(EntityType.FEED, clicksFeed);
         inputFeeds.add(clicksFeed);
-        verifyEntityWasAddedToGraph(clicksFeed.getName(),
-                EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(clicksFeed.getName(), RelationshipType.FEED_ENTITY);
         Assert.assertEquals(getVerticesCount(service.getGraph()), 9); // feed and financial vertex
         Assert.assertEquals(getEdgesCount(service.getGraph()), 11); // +5 = cluster + user + 2Group + Tag
 
@@ -169,16 +168,16 @@ public class MetadataMappingServiceTest {
                 Storage.TYPE.FILESYSTEM, "/falcon/imp-click-join1/${YEAR}${MONTH}${DAY}");
         configStore.publish(EntityType.FEED, join1Feed);
         outputFeeds.add(join1Feed);
-        verifyEntityWasAddedToGraph(join1Feed.getName(),
-                EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(join1Feed.getName(), RelationshipType.FEED_ENTITY);
         Assert.assertEquals(getVerticesCount(service.getGraph()), 12); // + 3 = 1 feed and 2 groups
-        Assert.assertEquals(getEdgesCount(service.getGraph()), 16); // +5 = cluster + user + Group + 2Tags
+        Assert.assertEquals(getEdgesCount(service.getGraph()), 16); // +5 = cluster + user +
+        // Group + 2Tags
 
         Feed join2Feed = buildFeed("imp-click-join2", clusterEntity, "classified-as=Secure,classified-as=Financial",
                 "reporting,bi", Storage.TYPE.FILESYSTEM, "/falcon/imp-click-join2/${YEAR}${MONTH}${DAY}");
         configStore.publish(EntityType.FEED, join2Feed);
         outputFeeds.add(join2Feed);
-        verifyEntityWasAddedToGraph(join2Feed.getName(), EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(join2Feed.getName(), RelationshipType.FEED_ENTITY);
 
         Assert.assertEquals(getVerticesCount(service.getGraph()), 13); // +1 feed
         // +6 = user + 2tags + 2Groups + Cluster
@@ -200,7 +199,7 @@ public class MetadataMappingServiceTest {
 
         configStore.publish(EntityType.PROCESS, processEntity);
 
-        verifyEntityWasAddedToGraph(processEntity.getName(), EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph(processEntity.getName(), RelationshipType.PROCESS_ENTITY);
         verifyProcessEntityEdges();
 
         // +2 = 1 process + 1 tag
@@ -211,7 +210,7 @@ public class MetadataMappingServiceTest {
 
     @Test (dependsOnMethods = "testOnAddProcessEntity")
     public void testOnAdd() throws Exception {
-        verifyEntityGraph(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE, "Secure");
+        verifyEntityGraph(RelationshipType.FEED_ENTITY, "Secure");
     }
 
     @Test(dependsOnMethods = "testOnAdd")
@@ -226,7 +225,7 @@ public class MetadataMappingServiceTest {
 
         debug(service.getGraph());
         GraphUtils.dump(service.getGraph());
-        verifyLineageGraph(InstanceRelationshipGraphBuilder.FEED_INSTANCE_TYPE);
+        verifyLineageGraph(RelationshipType.FEED_INSTANCE.getName());
 
         // +6 = 1 process, 2 inputs,2 outputs
         Assert.assertEquals(getVerticesCount(service.getGraph()), 20);
@@ -243,7 +242,7 @@ public class MetadataMappingServiceTest {
         // cannot modify cluster, adding a new cluster
         bcpCluster = buildCluster("bcp-cluster", "east-coast", "classification=bcp");
         configStore.publish(EntityType.CLUSTER, bcpCluster);
-        verifyEntityWasAddedToGraph("bcp-cluster", EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+        verifyEntityWasAddedToGraph("bcp-cluster", RelationshipType.CLUSTER_ENTITY);
 
         Assert.assertEquals(getVerticesCount(service.getGraph()), 23); // +3 = cluster, colo, tag
         // +2 edges to above, no user but only to colo and new tag
@@ -277,11 +276,10 @@ public class MetadataMappingServiceTest {
     }
 
     private void verifyUpdatedEdges(Feed newFeed) {
-        Vertex feedVertex = getEntityVertex(newFeed.getName(),
-                EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        Vertex feedVertex = getEntityVertex(newFeed.getName(), RelationshipType.FEED_ENTITY);
 
         // groups
-        Edge edge = feedVertex.getEdges(Direction.OUT, RelationshipGraphBuilder.GROUPS_LABEL).iterator().next();
+        Edge edge = feedVertex.getEdges(Direction.OUT, RelationshipLabel.GROUPS.getName()).iterator().next();
         Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), "reporting");
 
         // tags
@@ -292,8 +290,7 @@ public class MetadataMappingServiceTest {
 
         // new cluster
         List<String> actual = new ArrayList<String>();
-        for (Edge clusterEdge : feedVertex.getEdges(
-                Direction.OUT, RelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL)) {
+        for (Edge clusterEdge : feedVertex.getEdges(Direction.OUT, RelationshipLabel.FEED_CLUSTER_EDGE.getName())) {
             actual.add(clusterEdge.getVertex(Direction.IN).<String>getProperty("name"));
         }
         Assert.assertTrue(actual.containsAll(Arrays.asList("primary-cluster", "bcp-cluster")),
@@ -320,22 +317,20 @@ public class MetadataMappingServiceTest {
     }
 
     private void verifyUpdatedEdges(Process newProcess) {
-        Vertex processVertex = getEntityVertex(newProcess.getName(),
-                EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+        Vertex processVertex = getEntityVertex(newProcess.getName(), RelationshipType.PROCESS_ENTITY);
 
         // cluster
         Edge edge = processVertex.getEdges(Direction.OUT,
-                RelationshipGraphBuilder.PROCESS_CLUSTER_EDGE_LABEL).iterator().next();
+                RelationshipLabel.PROCESS_CLUSTER_EDGE.getName()).iterator().next();
         Assert.assertEquals(edge.getVertex(Direction.IN).getProperty("name"), bcpCluster.getName());
 
         // inputs
-        edge = processVertex.getEdges(Direction.IN,
-                RelationshipGraphBuilder.FEED_PROCESS_EDGE_LABEL).iterator().next();
+        edge = processVertex.getEdges(Direction.IN, RelationshipLabel.FEED_PROCESS_EDGE.getName()).iterator().next();
         Assert.assertEquals(edge.getVertex(Direction.OUT).getProperty("name"),
                 newProcess.getInputs().getInputs().get(0).getFeed());
 
         // outputs
-        for (Edge e : processVertex.getEdges(Direction.OUT, RelationshipGraphBuilder.PROCESS_FEED_EDGE_LABEL)) {
+        for (Edge e : processVertex.getEdges(Direction.OUT, RelationshipLabel.PROCESS_FEED_EDGE.getName())) {
             Assert.fail("there should not be any edges to output feeds" + e);
         }
     }
@@ -453,72 +448,72 @@ public class MetadataMappingServiceTest {
         outputs.getOutputs().add(output);
     }
 
-    private void verifyEntityWasAddedToGraph(String entityName, String entityType) {
+    private void verifyEntityWasAddedToGraph(String entityName, RelationshipType entityType) {
         Vertex entityVertex = getEntityVertex(entityName, entityType);
         Assert.assertNotNull(entityVertex);
         verifyEntityProperties(entityVertex, entityName, entityType);
     }
 
-    private void verifyEntityProperties(Vertex entityVertex, String entityName, String entityType) {
-        Assert.assertEquals(entityName, entityVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
-        Assert.assertEquals(entityType, entityVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
-        Assert.assertNotNull(entityVertex.getProperty(EntityRelationshipGraphBuilder.TIMESTAMP_PROPERTY_KEY));
+    private void verifyEntityProperties(Vertex entityVertex, String entityName, RelationshipType entityType) {
+        Assert.assertEquals(entityName, entityVertex.getProperty(RelationshipProperty.NAME.getName()));
+        Assert.assertEquals(entityType.getName(), entityVertex.getProperty(RelationshipProperty.TYPE.getName()));
+        Assert.assertNotNull(entityVertex.getProperty(RelationshipProperty.TIMESTAMP.getName()));
     }
 
     private void verifyClusterEntityEdges() {
         Vertex clusterVertex = getEntityVertex(CLUSTER_ENTITY_NAME,
-                EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+                RelationshipType.CLUSTER_ENTITY);
 
         // verify edge to user vertex
-        verifyVertexForEdge(clusterVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
-                FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+        verifyVertexForEdge(clusterVertex, Direction.OUT, RelationshipLabel.USER.getName(),
+                FALCON_USER, RelationshipType.USER.getName());
         // verify edge to colo vertex
-        verifyVertexForEdge(clusterVertex, Direction.OUT, EntityRelationshipGraphBuilder.CLUSTER_COLO_LABEL,
-                COLO_NAME, EntityRelationshipGraphBuilder.COLO_TYPE);
+        verifyVertexForEdge(clusterVertex, Direction.OUT, RelationshipLabel.CLUSTER_COLO.getName(),
+                COLO_NAME, RelationshipType.COLO.getName());
         // verify edge to tags vertex
         verifyVertexForEdge(clusterVertex, Direction.OUT, "classification",
-                "production", EntityRelationshipGraphBuilder.TAGS_TYPE);
+                "production", RelationshipType.TAGS.getName());
     }
 
     private void verifyFeedEntityEdges(String feedName) {
-        Vertex feedVertex = getEntityVertex(feedName, EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE);
+        Vertex feedVertex = getEntityVertex(feedName, RelationshipType.FEED_ENTITY);
 
         // verify edge to cluster vertex
-        verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL,
-                CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+        verifyVertexForEdge(feedVertex, Direction.OUT, RelationshipLabel.FEED_CLUSTER_EDGE.getName(),
+                CLUSTER_ENTITY_NAME, RelationshipType.CLUSTER_ENTITY.getName());
         // verify edge to user vertex
-        verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
-                FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+        verifyVertexForEdge(feedVertex, Direction.OUT, RelationshipLabel.USER.getName(),
+                FALCON_USER, RelationshipType.USER.getName());
         // verify edge to tags vertex
         verifyVertexForEdge(feedVertex, Direction.OUT, "classified-as",
-                "Secure", EntityRelationshipGraphBuilder.TAGS_TYPE);
+                "Secure", RelationshipType.TAGS.getName());
         // verify edge to group vertex
-        verifyVertexForEdge(feedVertex, Direction.OUT, EntityRelationshipGraphBuilder.GROUPS_LABEL,
-                "analytics", EntityRelationshipGraphBuilder.GROUPS_TYPE);
+        verifyVertexForEdge(feedVertex, Direction.OUT, RelationshipLabel.GROUPS.getName(),
+                "analytics", RelationshipType.GROUPS.getName());
     }
 
     private void verifyProcessEntityEdges() {
         Vertex processVertex = getEntityVertex(PROCESS_ENTITY_NAME,
-                EntityRelationshipGraphBuilder.PROCESS_ENTITY_TYPE);
+                RelationshipType.PROCESS_ENTITY);
 
         // verify edge to cluster vertex
-        verifyVertexForEdge(processVertex, Direction.OUT, EntityRelationshipGraphBuilder.FEED_CLUSTER_EDGE_LABEL,
-                CLUSTER_ENTITY_NAME, EntityRelationshipGraphBuilder.CLUSTER_ENTITY_TYPE);
+        verifyVertexForEdge(processVertex, Direction.OUT, RelationshipLabel.FEED_CLUSTER_EDGE.getName(),
+                CLUSTER_ENTITY_NAME, RelationshipType.CLUSTER_ENTITY.getName());
         // verify edge to user vertex
-        verifyVertexForEdge(processVertex, Direction.OUT, EntityRelationshipGraphBuilder.USER_LABEL,
-                FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
+        verifyVertexForEdge(processVertex, Direction.OUT, RelationshipLabel.USER.getName(),
+                FALCON_USER, RelationshipType.USER.getName());
         // verify edge to tags vertex
         verifyVertexForEdge(processVertex, Direction.OUT, "classified-as",
-                "Critical", EntityRelationshipGraphBuilder.TAGS_TYPE);
+                "Critical", RelationshipType.TAGS.getName());
 
         // verify edges to inputs
         List<String> actual = new ArrayList<String>();
         for (Edge edge : processVertex.getEdges(Direction.IN,
-                EntityRelationshipGraphBuilder.FEED_PROCESS_EDGE_LABEL)) {
+                RelationshipLabel.FEED_PROCESS_EDGE.getName())) {
             Vertex outVertex = edge.getVertex(Direction.OUT);
-            Assert.assertEquals(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE,
-                    outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
-            actual.add(outVertex.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+            Assert.assertEquals(RelationshipType.FEED_ENTITY.getName(),
+                    outVertex.getProperty(RelationshipProperty.TYPE.getName()));
+            actual.add(outVertex.<String>getProperty(RelationshipProperty.NAME.getName()));
         }
 
         Assert.assertTrue(actual.containsAll(Arrays.asList("impression-feed", "clicks-feed")),
@@ -527,20 +522,20 @@ public class MetadataMappingServiceTest {
         actual.clear();
         // verify edges to outputs
         for (Edge edge : processVertex.getEdges(Direction.OUT,
-                EntityRelationshipGraphBuilder.PROCESS_FEED_EDGE_LABEL)) {
+                RelationshipLabel.PROCESS_FEED_EDGE.getName())) {
             Vertex outVertex = edge.getVertex(Direction.IN);
-            Assert.assertEquals(EntityRelationshipGraphBuilder.FEED_ENTITY_TYPE,
-                    outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY));
-            actual.add(outVertex.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+            Assert.assertEquals(RelationshipType.FEED_ENTITY.getName(),
+                    outVertex.getProperty(RelationshipProperty.TYPE.getName()));
+            actual.add(outVertex.<String>getProperty(RelationshipProperty.NAME.getName()));
         }
         Assert.assertTrue(actual.containsAll(Arrays.asList("imp-click-join1", "imp-click-join2")),
                 "Actual does not contain expected: " + actual);
     }
 
-    private Vertex getEntityVertex(String entityName, String entityType) {
+    private Vertex getEntityVertex(String entityName, RelationshipType entityType) {
         GraphQuery entityQuery = getQuery()
-                .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, entityName)
-                .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, entityType);
+                .has(RelationshipProperty.NAME.getName(), entityName)
+                .has(RelationshipProperty.TYPE.getName(), entityType.getName());
         Iterator<Vertex> iterator = entityQuery.vertices().iterator();
         Assert.assertTrue(iterator.hasNext());
 
@@ -555,38 +550,38 @@ public class MetadataMappingServiceTest {
         for (Edge edge : fromVertex.getEdges(direction, label)) {
             Vertex outVertex = edge.getVertex(Direction.IN);
             Assert.assertEquals(
-                    outVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY), expectedName);
+                    outVertex.getProperty(RelationshipProperty.NAME.getName()), expectedName);
             Assert.assertEquals(
-                    outVertex.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY), expectedType);
+                    outVertex.getProperty(RelationshipProperty.TYPE.getName()), expectedType);
         }
     }
 
-    private void verifyEntityGraph(String feedType, String classification) {
+    private void verifyEntityGraph(RelationshipType feedType, String classification) {
         // feeds owned by a user
-        List<String> feedNamesOwnedByUser = getFeedsOwnedByAUser(feedType);
+        List<String> feedNamesOwnedByUser = getFeedsOwnedByAUser(feedType.getName());
         Assert.assertEquals(feedNamesOwnedByUser,
                 Arrays.asList("impression-feed", "clicks-feed", "imp-click-join1", "imp-click-join2"));
 
         // feeds classified as secure
-        verifyFeedsClassifiedAsSecure(feedType,
+        verifyFeedsClassifiedAsSecure(feedType.getName(),
                 Arrays.asList("impression-feed", "clicks-feed", "imp-click-join2"));
 
         // feeds owned by a user and classified as secure
-        verifyFeedsOwnedByUserAndClassification(feedType, classification,
+        verifyFeedsOwnedByUserAndClassification(feedType.getName(), classification,
                 Arrays.asList("impression-feed", "clicks-feed", "imp-click-join2"));
     }
 
     private List<String> getFeedsOwnedByAUser(String feedType) {
         GraphQuery userQuery = getQuery()
-                .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, FALCON_USER)
-                .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, EntityRelationshipGraphBuilder.USER_TYPE);
+                .has(RelationshipProperty.NAME.getName(), FALCON_USER)
+                .has(RelationshipProperty.TYPE.getName(), RelationshipType.USER.getName());
 
         List<String> feedNames = new ArrayList<String>();
         for (Vertex userVertex : userQuery.vertices()) {
-            for (Vertex feed : userVertex.getVertices(Direction.IN, EntityRelationshipGraphBuilder.USER_LABEL)) {
-                if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+            for (Vertex feed : userVertex.getVertices(Direction.IN, RelationshipLabel.USER.getName())) {
+                if (feed.getProperty(RelationshipProperty.TYPE.getName()).equals(feedType)) {
                     System.out.println(FALCON_USER + " owns -> " + GraphUtils.vertexString(feed));
-                    feedNames.add(feed.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+                    feedNames.add(feed.<String>getProperty(RelationshipProperty.NAME.getName()));
                 }
             }
         }
@@ -596,15 +591,15 @@ public class MetadataMappingServiceTest {
 
     private void verifyFeedsClassifiedAsSecure(String feedType, List<String> expected) {
         GraphQuery classQuery = getQuery()
-                .has(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY, "Secure")
-                .has(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY, EntityRelationshipGraphBuilder.TAGS_TYPE);
+                .has(RelationshipProperty.NAME.getName(), "Secure")
+                .has(RelationshipProperty.TYPE.getName(), RelationshipType.TAGS.getName());
 
         List<String> actual = new ArrayList<String>();
         for (Vertex feedVertex : classQuery.vertices()) {
             for (Vertex feed : feedVertex.getVertices(Direction.BOTH, "classified-as")) {
-                if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+                if (feed.getProperty(RelationshipProperty.TYPE.getName()).equals(feedType)) {
                     System.out.println(" Secure classification -> " + GraphUtils.vertexString(feed));
-                    actual.add(feed.<String>getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY));
+                    actual.add(feed.<String>getProperty(RelationshipProperty.NAME.getName()));
                 }
             }
         }
@@ -615,13 +610,13 @@ public class MetadataMappingServiceTest {
     private void verifyFeedsOwnedByUserAndClassification(String feedType, String classification,
                                                          List<String> expected) {
         List<String> actual = new ArrayList<String>();
-        Vertex userVertex = getEntityVertex(FALCON_USER, EntityRelationshipGraphBuilder.USER_TYPE);
-        for (Vertex feed : userVertex.getVertices(Direction.IN, EntityRelationshipGraphBuilder.USER_LABEL)) {
-            if (feed.getProperty(EntityRelationshipGraphBuilder.TYPE_PROPERTY_KEY).equals(feedType)) {
+        Vertex userVertex = getEntityVertex(FALCON_USER, RelationshipType.USER);
+        for (Vertex feed : userVertex.getVertices(Direction.IN, RelationshipLabel.USER.getName())) {
+            if (feed.getProperty(RelationshipProperty.TYPE.getName()).equals(feedType)) {
                 for (Vertex classVertex : feed.getVertices(Direction.OUT, "classified-as")) {
-                    if (classVertex.getProperty(EntityRelationshipGraphBuilder.NAME_PROPERTY_KEY)
+                    if (classVertex.getProperty(RelationshipProperty.NAME.getName())
                             .equals(classification)) {
-                        actual.add(feed.<String>getProperty(RelationshipGraphBuilder.NAME_PROPERTY_KEY));
+                        actual.add(feed.<String>getProperty(RelationshipProperty.NAME.getName()));
                         System.out.println(classification + " feed owned by falcon-user -> "
                                 + GraphUtils.vertexString(feed));
                     }
@@ -661,9 +656,8 @@ public class MetadataMappingServiceTest {
         Iterator<Vertex> vertices = graph.getVertices("name", "impression-feed/2014-01-01T00:00Z").iterator();
         Assert.assertTrue(vertices.hasNext());
         Vertex feedInstanceVertex = vertices.next();
-        Assert.assertEquals(
-                feedInstanceVertex.getProperty(RelationshipGraphBuilder.TYPE_PROPERTY_KEY),
-                InstanceRelationshipGraphBuilder.FEED_INSTANCE_TYPE);
+        Assert.assertEquals(feedInstanceVertex.getProperty(RelationshipProperty.TYPE.getName()),
+                RelationshipType.FEED_INSTANCE.getName());
 
         Object vertexId = feedInstanceVertex.getId();
         Vertex vertexById = graph.getVertex(vertexId);

http://git-wip-us.apache.org/repos/asf/incubator-falcon/blob/4e8065c5/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
index 64a1d77..a1aeca5 100644
--- a/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
+++ b/prism/src/main/java/org/apache/falcon/resource/metadata/LineageMetadataResource.java
@@ -28,6 +28,9 @@ import com.tinkerpop.blueprints.util.io.graphson.GraphSONMode;
 import com.tinkerpop.blueprints.util.io.graphson.GraphSONUtility;
 import org.apache.falcon.metadata.GraphUtils;
 import org.apache.falcon.metadata.MetadataMappingService;
+import org.apache.falcon.metadata.RelationshipLabel;
+import org.apache.falcon.metadata.RelationshipProperty;
+import org.apache.falcon.metadata.RelationshipType;
 import org.apache.falcon.service.Services;
 import org.apache.falcon.util.StartupProperties;
 import org.apache.log4j.Logger;
@@ -35,6 +38,7 @@ import org.codehaus.jettison.json.JSONArray;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
 
+import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
@@ -43,6 +47,8 @@ import javax.ws.rs.QueryParam;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -92,7 +98,7 @@ public class LineageMetadataResource {
     public Response serializeGraph() {
         checkIfMetadataMappingServiceIsEnabled();
         String file = StartupProperties.get().getProperty("falcon.graph.serialize.path")
-                + "lineage-graph-" + System.currentTimeMillis() + ".json";
+                + "/lineage-graph-" + System.currentTimeMillis() + ".json";
         LOG.info("Serialize Graph to: " + file);
         try {
             GraphUtils.dump(getGraph(), file);
@@ -137,13 +143,7 @@ public class LineageMetadataResource {
         checkIfMetadataMappingServiceIsEnabled();
         LOG.info("Get vertex for vertexId= " + vertexId);
         try {
-            Vertex vertex = getGraph().getVertex(vertexId);
-            if (vertex == null) {
-                String message = "Vertex with [" + vertexId + "] cannot be found.";
-                LOG.info(message);
-                throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
-                        .entity(JSONObject.quote(message)).build());
-            }
+            Vertex vertex = findVertex(vertexId);
 
             JSONObject response = new JSONObject();
             response.put(RESULTS, GraphSONUtility.jsonFromElement(
@@ -155,6 +155,118 @@ public class LineageMetadataResource {
         }
     }
 
+    private Vertex findVertex(String vertexId) {
+        Vertex vertex = getGraph().getVertex(vertexId);
+        if (vertex == null) {
+            String message = "Vertex with [" + vertexId + "] cannot be found.";
+            LOG.info(message);
+            throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
+                    .entity(JSONObject.quote(message)).build());
+        }
+
+        return vertex;
+    }
+
+    /**
+     * Get properties for a single vertex with a unique id.
+     * This is NOT a rexster API.
+     * <p/>
+     * GET http://host/graphs/lineage/vertices/properties/id
+     */
+    @GET
+    @Path("/vertices/properties/{id}")
+    @Produces({MediaType.APPLICATION_JSON})
+    public Response getVertexProperties(@PathParam("id") final String vertexId,
+                                        @DefaultValue("false") @QueryParam("relationships")
+                                        final String relationships) {
+        checkIfMetadataMappingServiceIsEnabled();
+        LOG.info("Get vertex for vertexId= " + vertexId);
+        try {
+            Vertex vertex = findVertex(vertexId);
+
+            Map<String, String> vertexProperties = getVertexProperties(vertex, Boolean.valueOf(relationships));
+
+            JSONObject response = new JSONObject();
+            response.put(RESULTS, new JSONObject(vertexProperties));
+            response.put(TOTAL_SIZE, vertexProperties.size());
+            return Response.ok(response).build();
+        } catch (JSONException e) {
+            throw new WebApplicationException(Response.status(Response.Status.INTERNAL_SERVER_ERROR)
+                    .entity(JSONObject.quote("An error occurred: " + e.getMessage())).build());
+        }
+    }
+
+    private Map<String, String> getVertexProperties(Vertex vertex, boolean captureRelationships) {
+        Map<String, String> vertexProperties = new HashMap<String, String>();
+        for (String key : vertex.getPropertyKeys()) {
+            vertexProperties.put(key, vertex.<String>getProperty(key));
+        }
+
+        RelationshipType vertexType = RelationshipType.fromString(
+                vertex.<String>getProperty(RelationshipProperty.TYPE.getName()));
+        // get the properties from relationships
+        if (captureRelationships && (vertexType == RelationshipType.FEED_INSTANCE
+                || vertexType == RelationshipType.PROCESS_INSTANCE)) {
+            for (Edge edge : vertex.getEdges(Direction.OUT)) {
+                Vertex toVertex = edge.getVertex(Direction.IN);
+                addRelationships(vertexType, toVertex, vertexProperties);
+            }
+        }
+
+        return vertexProperties;
+    }
+
+    private void addRelationships(RelationshipType fromVertexType, Vertex toVertex,
+                                  Map<String, String> vertexProperties) {
+        String value = toVertex.getProperty(RelationshipProperty.NAME.getName());
+        RelationshipType toVertexType = RelationshipType.fromString(
+                toVertex.<String>getProperty(RelationshipProperty.TYPE.getName()));
+
+        switch (toVertexType) {
+        case CLUSTER_ENTITY:
+            String key = fromVertexType == RelationshipType.FEED_INSTANCE
+                    ? RelationshipLabel.FEED_CLUSTER_EDGE.getName()
+                    : RelationshipLabel.PROCESS_CLUSTER_EDGE.getName();
+            vertexProperties.put(key, value);
+            break;
+
+        case USER:
+            vertexProperties.put(RelationshipLabel.USER.getName(), value);
+            break;
+
+        case FEED_ENTITY:
+            addEntityRelationships(toVertex, vertexProperties);
+            break;
+
+        case PROCESS_ENTITY:
+            addEntityRelationships(toVertex, vertexProperties);
+            break;
+
+        default:
+        }
+    }
+
+    private void addEntityRelationships(Vertex vertex, Map<String, String> vertexProperties) {
+        for (Edge edge : vertex.getEdges(Direction.OUT)) {
+            Vertex toVertex = edge.getVertex(Direction.IN);
+            String value = toVertex.getProperty(RelationshipProperty.NAME.getName());
+            RelationshipType toVertexType = RelationshipType.fromString(
+                    toVertex.<String>getProperty(RelationshipProperty.TYPE.getName()));
+
+            switch (toVertexType) {
+            case TAGS:
+                vertexProperties.put(edge.getLabel(), value);
+                break;
+
+            case GROUPS:
+                vertexProperties.put(RelationshipLabel.GROUPS.getName(), value);
+                break;
+
+            default:
+            }
+        }
+    }
+
     /**
      * Get a list of vertices matching a property key and a value.
      * <p/>
@@ -193,13 +305,7 @@ public class LineageMetadataResource {
         checkIfMetadataMappingServiceIsEnabled();
         LOG.info("Get vertex edges for vertexId= " + vertexId + ", direction= " + direction);
         try {
-            Vertex vertex = getGraph().getVertex(vertexId);
-            if (vertex == null) {
-                String message = "Vertex with [" + vertexId + "] cannot be found.";
-                LOG.info(message);
-                throw new WebApplicationException(Response.status(Response.Status.NOT_FOUND)
-                        .entity(JSONObject.quote(message)).build());
-            }
+            Vertex vertex = findVertex(vertexId);
 
             return getVertexEdges(vertex, direction);
 


Mime
View raw message