diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
index c136811..9151d48 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/DAG.java
@@ -273,6 +273,13 @@
 
   /**
    * Add an {@link Edge} connecting vertices in the DAG
+   *
+   * All edges within a DAG must be either named (created via
+   * {@link org.apache.tez.dag.api.Edge#create(Vertex, Vertex, EdgeProperty, String)}) or unnamed
+   * (created via {@link org.apache.tez.dag.api.Edge#create(Vertex, Vertex, EdgeProperty)}).
+   * If edges are named, all inbound edges to a vertex should have unique names. Likewise for outbound edges.
+   * A vertex can have an inbound edge that uses the same name as that used by an outbound edge.
+   *
    * @param edge The edge to be added
    * @return {@link DAG}
    */
@@ -571,8 +578,6 @@
     // check for valid vertices, duplicate vertex names,
     // and prepare for cycle detection
     Map<String, AnnotatedVertex> vertexMap = new HashMap<String, AnnotatedVertex>();
-    Map<Vertex, Set<String>> inboundVertexMap = new HashMap<Vertex, Set<String>>();
-    Map<Vertex, Set<String>> outboundVertexMap = new HashMap<Vertex, Set<String>>();
     for (Vertex v : vertices.values()) {
       if (vertexMap.containsKey(v.getName())) {
         throw new IllegalStateException("DAG contains multiple vertices"
@@ -581,33 +586,45 @@
       vertexMap.put(v.getName(), new AnnotatedVertex(v));
     }
 
-    Map<Vertex, List<Edge>> edgeMap = new HashMap<Vertex, List<Edge>>();
+    // named edge cannot be mixed with unnamed edge or group edge
+    Edge namedEdge = null, unnamedEdge = null;
+    for (Edge e : edges) {
+      if (e.getName() == null) {
+        unnamedEdge = e;
+      } else {
+        namedEdge = e;
+      }
+
+      if (namedEdge != null && !groupInputEdges.isEmpty()) {
+        throw new IllegalStateException("DAG shouldn't contains both named edge " + namedEdge
+          + " and group edge " + groupInputEdges.iterator().next());
+      }
+      if (namedEdge != null && unnamedEdge != null) {
+        throw new IllegalStateException("DAG shouldn't contains both named edge " + namedEdge
+          + " and unnamed edge " + unnamedEdge);
+      }
+    }
+
+    Map<Vertex, List<Edge>> inEdgeMap = new HashMap<>();
+    Map<Vertex, List<Edge>> outEdgeMap = new HashMap<>();
     for (Edge e : edges) {
       // Construct structure for cycle detection
       Vertex inputVertex = e.getInputVertex();
       Vertex outputVertex = e.getOutputVertex();      
-      List<Edge> edgeList = edgeMap.get(inputVertex);
-      if (edgeList == null) {
-        edgeList = new ArrayList<Edge>();
-        edgeMap.put(inputVertex, edgeList);
+
+      List<Edge> outEdgeList = outEdgeMap.get(inputVertex);
+      if (outEdgeList == null) {
+        outEdgeList = new ArrayList<Edge>();
+        outEdgeMap.put(inputVertex, outEdgeList);
       }
-      edgeList.add(e);
-      
-      // Construct map for Input name verification
-      Set<String> inboundSet = inboundVertexMap.get(outputVertex);
-      if (inboundSet == null) {
-        inboundSet = new HashSet<String>();
-        inboundVertexMap.put(outputVertex, inboundSet);
+      outEdgeList.add(e);
+
+      List<Edge> inEdgeList = inEdgeMap.get(outputVertex);
+      if (inEdgeList == null) {
+        inEdgeList = new ArrayList<Edge>();
+        inEdgeMap.put(outputVertex, inEdgeList);
       }
-      inboundSet.add(inputVertex.getName());
-      
-      // Construct map for Output name verification
-      Set<String> outboundSet = outboundVertexMap.get(inputVertex);
-      if (outboundSet == null) {
-        outboundSet = new HashSet<String>();
-        outboundVertexMap.put(inputVertex, outboundSet);
-      }
-      outboundSet.add(outputVertex.getName());
+      inEdgeList.add(e);
     }
 
     // check input and output names don't collide with vertex names
@@ -633,29 +650,54 @@
     }
 
     // Check for valid InputNames
-    for (Entry<Vertex, Set<String>> entry : inboundVertexMap.entrySet()) {
+    for (Entry<Vertex, List<Edge>> entry : inEdgeMap.entrySet()) {
       Vertex vertex = entry.getKey();
+      Set<String> inputs = new HashSet<>();
+
+      for (Edge edge : entry.getValue()) {
+        String name = edge.getName();
+        if (name == null) {
+          name = edge.getInputVertex().getName();
+        }
+        if (inputs.contains(name)) {
+          throw new IllegalStateException("Vertex: " + vertex.getName() + " contains multiple " +
+            "incoming edges with name " + name);
+        }
+        inputs.add(name);
+      }
       for (RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor> 
            input : vertex.getInputs()) {
-        if (entry.getValue().contains(input.getName())) {
+        if (inputs.contains(input.getName())) {
           throw new IllegalStateException("Vertex: "
               + vertex.getName()
-              + " contains an incoming vertex and Input with the same name: "
-              + input.getName());
+              + " contains an incoming " + (namedEdge != null ? "edge" : "vertex")
+              + " and Input with the same name: " + input.getName());
         }
       }
     }
 
-    // Check for valid OutputNames
-    for (Entry<Vertex, Set<String>> entry : outboundVertexMap.entrySet()) {
+    for (Entry<Vertex, List<Edge>> entry : outEdgeMap.entrySet()) {
       Vertex vertex = entry.getKey();
-      for (RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> 
-            output : vertex.getOutputs()) {
-        if (entry.getValue().contains(output.getName())) {
+      Set<String> outputs = new HashSet<>();
+
+      for (Edge edge : entry.getValue()) {
+        String name = edge.getName();
+        if (name == null) {
+          name = edge.getOutputVertex().getName();
+        }
+        if (outputs.contains(name)) {
+          throw new IllegalStateException("Vertex: " + vertex.getName() + " contains multiple " +
+            "outgoing edges with name " + name);
+        }
+        outputs.add(name);
+      }
+      for (RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor>
+        output : vertex.getOutputs()) {
+        if (outputs.contains(output.getName())) {
           throw new IllegalStateException("Vertex: "
-              + vertex.getName()
-              + " contains an outgoing vertex and Output with the same name: "
-              + output.getName());
+            + vertex.getName()
+            + " contains an outgoing " + (namedEdge != null ? "edge" : "vertex")
+            + " and Output with the same name: " + output.getName());
         }
       }
     }
@@ -666,7 +708,7 @@
     // When additional inputs are supported, this can be chceked easily (and early)
     // within the addInput / addOutput call itself.
 
-    Deque<String> topologicalVertexStack = detectCycles(edgeMap, vertexMap);
+    Deque<String> topologicalVertexStack = detectCycles(outEdgeMap, vertexMap);
 
     checkAndInferOneToOneParallelism();
 
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java b/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
index db509f7..794d88d 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/Edge.java
@@ -24,7 +24,7 @@
  * Edge defines the connection between a producer and consumer vertex in the DAG.
  * @link {@link EdgeProperty} defines the relationship between them. The producer
  * vertex provides input to the edge and the consumer vertex reads output from the 
- * edge.
+ * edge. Edge could be either named or not.
  * 
  */
 @Public
@@ -33,13 +33,16 @@
   private final Vertex inputVertex;
   private final Vertex outputVertex;
   private final EdgeProperty edgeProperty;
+  private final String name;
 
   private Edge(Vertex inputVertex,
                Vertex outputVertex,
-               EdgeProperty edgeProperty) {
+               EdgeProperty edgeProperty,
+               String name) {
     this.inputVertex = inputVertex;
     this.outputVertex = outputVertex;
     this.edgeProperty = edgeProperty;
+    this.name = name;
   }
 
 
@@ -57,7 +60,25 @@
   public static Edge create(Vertex inputVertex,
                             Vertex outputVertex,
                             EdgeProperty edgeProperty) {
-    return new Edge(inputVertex, outputVertex, edgeProperty);
+    return new Edge(inputVertex, outputVertex, edgeProperty, null);
+  }
+
+  /**
+   * Creates an edge with specified name between the specified vertices.
+   *
+   * InputVertex(EdgeInput) ----- Edge ----- OutputVertex(EdgeOutput)]
+   *
+   * @param inputVertex the vertex which generates data to the edge.
+   * @param outputVertex the vertex which consumes data from the edge
+   * @param edgeProperty {@link org.apache.tez.dag.api.EdgeProperty} associated with this edge
+   * @param name name of edge
+   * @return the {@link org.apache.tez.dag.api.Edge}
+   */
+  public static Edge create(Vertex inputVertex,
+                            Vertex outputVertex,
+                            EdgeProperty edgeProperty,
+                            String name) {
+    return new Edge(inputVertex, outputVertex, edgeProperty, name);
   }
 
   /**
@@ -83,6 +104,14 @@
   public EdgeProperty getEdgeProperty() {
     return edgeProperty;
   }
+
+  /**
+   * The name of this edge (or null if edge has no name)
+   * @return edge name or null
+   */
+  public String getName() {
+    return name;
+  }
   
   /*
    * Used to identify the edge in the configuration
@@ -95,39 +124,29 @@
  
   @Override
   public String toString() {
-    return inputVertex + " -> " + outputVertex + " (" + edgeProperty + ")";
+    return "{" + name + " : " + inputVertex + " -> " + outputVertex + " " + edgeProperty + "}";
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    if (this == other) return true;
+    if (other == null || getClass() != other.getClass()) return false;
+
+    Edge edge = (Edge) other;
+
+    if (inputVertex != null ? !inputVertex.equals(edge.inputVertex) : edge.inputVertex != null)
+      return false;
+    if (outputVertex != null ? !outputVertex.equals(edge.outputVertex) : edge.outputVertex != null)
+      return false;
+    return name != null ? name.equals(edge.name) : edge.name == null;
+
   }
 
   @Override
   public int hashCode() {
-    final int prime = 31;
-    int result = 1;
-    result = prime * result
-        + ((inputVertex == null) ? 0 : inputVertex.hashCode());
-    result = prime * result
-        + ((outputVertex == null) ? 0 : outputVertex.hashCode());
+    int result = inputVertex != null ? inputVertex.hashCode() : 0;
+    result = 31 * result + (outputVertex != null ? outputVertex.hashCode() : 0);
+    result = 31 * result + (name != null ? name.hashCode() : 0);
     return result;
   }
-
-  @Override
-  public boolean equals(Object obj) {
-    if (this == obj)
-      return true;
-    if (obj == null)
-      return false;
-    if (getClass() != obj.getClass())
-      return false;
-    Edge other = (Edge) obj;
-    if (inputVertex == null) {
-      if (other.inputVertex != null)
-        return false;
-    } else if (!inputVertex.equals(other.inputVertex))
-      return false;
-    if (outputVertex == null) {
-      if (other.outputVertex != null)
-        return false;
-    } else if (!outputVertex.equals(other.outputVertex))
-      return false;
-    return true;
-  }
 }
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java b/tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
index 07fb2c1..3723433 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/EdgeProperty.java
@@ -243,10 +243,10 @@
   
   @Override
   public String toString() {
-    return "{ " + dataMovementType + " : " + inputDescriptor.getClassName()
+    return "(" + dataMovementType + " : " + inputDescriptor.getClassName()
         + " >> " + dataSourceType + " >> " + outputDescriptor.getClassName()
         + " >> " + (edgeManagerDescriptor == null ? "NullEdgeManager" : edgeManagerDescriptor.getClassName())
-        + " }";
+        + ")";
   }
   
 }
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAG.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAG.java
index 05c4e30..691b4c1 100644
--- a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAG.java
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAG.java
@@ -88,6 +88,32 @@
   }
 
   @Test(timeout = 5000)
+  public void testAddDuplicatedNamedEdge() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+    Edge edge2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+
+    DAG dag = DAG.create("testDAG").addVertex(v1).addVertex(v2).addEdge(edge1);
+
+    try {
+      dag.addEdge(edge2);
+      Assert.fail("should fail it due to duplicate named edges");
+    } catch (Exception e) {
+      Assert.assertTrue(e.getMessage().contains(edge1 + " already defined"));
+    }
+  }
+
+  @Test(timeout = 5000)
   public void testDuplicatedVertexGroup() {
     Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
         dummyTaskCount, dummyTaskResource);
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
index 5706542..720820d 100644
--- a/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestDAGVerify.java
@@ -1191,4 +1191,253 @@
       Assert.assertTrue(e.getMessage().contains("There is conflicting local resource"));
     }
   }
+
+  @Test(timeout = 5000)
+  public void testNamedEdge() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+
+    DAG dag = DAG.create("testDAG").addVertex(v1).addVertex(v2).addEdge(edge1);
+    dag.verify();
+
+    Edge edge2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge2");
+
+    dag.addEdge(edge2).verify();
+  }
+
+  @Test(timeout = 5000)
+  public void testNamedEdgeMixedWithUnnamedEdge() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+    Edge edge2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")));
+
+    DAG dag = DAG.create("testDAG").addVertex(v1).addVertex(v2).addEdge(edge1).addEdge(edge2);
+
+    try {
+      dag.verify();
+      Assert.fail("should fail it because DAG has both named and unnamed edge");
+    } catch (Exception e) {
+      Assert.assertTrue(
+        e.getMessage().contains(
+          "DAG shouldn't contains both named edge " + edge1 + " and unnamed edge " + edge2));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testNamedEdgeWithGroupEdge() {
+    Vertex v1 = Vertex.create("v1",
+      ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2",
+      ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v3 = Vertex.create("v3",
+      ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    DAG dag = DAG.create("testDag");
+    dag.addVertex(v1);
+    dag.addVertex(v2);
+    dag.addVertex(v3);
+    String groupName1 = "uv12";
+    VertexGroup uv12 = dag.createVertexGroup(groupName1, v1, v2);
+
+    GroupInputEdge e1 = GroupInputEdge.create(uv12, v3,
+      EdgeProperty.create(DataMovementType.SCATTER_GATHER,
+        DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+        OutputDescriptor.create("dummy output class"),
+        InputDescriptor.create("dummy input class")),
+      InputDescriptor.create("dummy input class"));
+
+    Edge e2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.CONCURRENT, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "e2");
+
+    dag.addEdge(e1).addEdge(e2);
+    try {
+      dag.verify();
+      Assert.fail("should fail it because DAG has both named edge and group edge");
+    } catch (Exception e) {
+      Assert.assertTrue(
+        e.getMessage().contains("DAG shouldn't contains both named edge " + e2 + " and group edge "
+          + e1));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testInNamedEdgeCollide() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v3 = Vertex.create("v3", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    Edge edge1 = Edge.create(v1, v3, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+    Edge edge2 = Edge.create(v2, v3, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+
+    DAG dag =
+      DAG.create("testDAG").addVertex(v1).addVertex(v2).addVertex(v3).addEdge(edge1).addEdge(edge2);
+
+    try {
+      dag.verify();
+      Assert.fail("should fail it because v3 gets multiple incoming edges with same name");
+    } catch (Exception e) {
+      Assert.assertTrue(
+        e.getMessage().contains("v3 contains multiple incoming edges with name Edge1"));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testOutNamedEdgeCollide() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v3 = Vertex.create("v3", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    Edge edge1 = Edge.create(v1, v3, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+    Edge edge2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+
+    DAG dag =
+      DAG.create("testDAG").addVertex(v1).addVertex(v2).addVertex(v3).addEdge(edge1).addEdge(edge2);
+
+    try {
+      dag.verify();
+      Assert.fail("should fail it because v3 gets multiple outgoing edges with same name");
+    } catch (Exception e) {
+      Assert.assertTrue(
+        e.getMessage().contains("v1 contains multiple outgoing edges with name Edge1"));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testInEdgeOutEdgeWithSameName() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v3 = Vertex.create("v3", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+    Edge edge2 = Edge.create(v2, v3, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "Edge1");
+
+    DAG.create("testDAG")
+      .addVertex(v1).addVertex(v2).addVertex(v3)
+      .addEdge(edge1).addEdge(edge2)
+      .verify();
+  }
+
+  @Test(timeout = 5000)
+  public void testNamedEdgeCollideWithRootInput() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    v2.addDataSource("input",
+      DataSourceDescriptor.create(InputDescriptor.create("input"), null, null));
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "input");
+
+    DAG dag = DAG.create("testDag").addVertex(v1).addVertex(v2).addEdge(edge1);
+
+    try {
+      dag.verify();
+      Assert.fail("should fail it because v2 get incoming edge and input with same name");
+    } catch (Exception e) {
+      Assert.assertTrue(
+        e.getMessage().contains("v2 contains an incoming edge and Input with the same name: input"));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testNamedEdgeCollideWithLeafOutput() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    v1.addDataSink("output",
+      DataSinkDescriptor.create(OutputDescriptor.create("output"), null, null));
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "output");
+
+    DAG dag = DAG.create("testDag").addVertex(v1).addVertex(v2).addEdge(edge1);
+
+    try {
+      dag.verify();
+      Assert.fail("should fail it because v2 get outgoing edge and output with same name");
+    } catch (Exception e) {
+      Assert.assertTrue(e.getMessage().contains(
+        "v1 contains an outgoing edge and Output with the same name: output"));
+    }
+  }
+
+  @Test(timeout = 5000)
+  public void testNamedEdgeUsingVertexName() {
+    Vertex v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+    Vertex v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"),
+      dummyTaskCount, dummyTaskResource);
+
+    Edge edge1 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "v1");
+    Edge edge2 = Edge.create(v1, v2, EdgeProperty.create(
+      DataMovementType.SCATTER_GATHER, DataSourceType.PERSISTED,
+      SchedulingType.SEQUENTIAL, OutputDescriptor.create("output"),
+      InputDescriptor.create("input")), "v2");
+
+    DAG dag = DAG.create("testDag").addVertex(v1).addVertex(v2).addEdge(edge1).addEdge(edge2);
+    dag.verify();
+  }
 }
diff --git a/tez-api/src/test/java/org/apache/tez/dag/api/TestEdge.java b/tez-api/src/test/java/org/apache/tez/dag/api/TestEdge.java
new file mode 100644
index 0000000..61f4fbc
--- /dev/null
+++ b/tez-api/src/test/java/org/apache/tez/dag/api/TestEdge.java
@@ -0,0 +1,75 @@
+/**
+ * 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.tez.dag.api;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestEdge {
+  Vertex v1, v2;
+  EdgeProperty edgeProperty;
+  Set<Edge> set;
+
+  @Before
+  public void setup() {
+    v1 = Vertex.create("v1", ProcessorDescriptor.create("Processor"));
+    v2 = Vertex.create("v2", ProcessorDescriptor.create("Processor"));
+    edgeProperty = EdgeProperty.create(EdgeProperty.DataMovementType.SCATTER_GATHER,
+      EdgeProperty.DataSourceType.PERSISTED, EdgeProperty.SchedulingType.CONCURRENT,
+      OutputDescriptor.create("output"), InputDescriptor.create("input"));
+    set = new HashSet<>();
+  }
+
+  @Test(timeout = 5000)
+  public void testHashAndEqualsUnnamed() {
+    // edges without name but everything else same are equal and have same hash
+    Edge e1 = Edge.create(v1, v2, edgeProperty);
+    Edge e2 = Edge.create(v1, v2, edgeProperty);
+    assertEquals(e1, e2);
+    set.add(e1);
+    assertTrue(set.contains(e2));
+  }
+
+  @Test(timeout = 5000)
+  public void testHashAndEqualsNamed() {
+    // edges with everything same including name are equal and have same hash
+    Edge e1 = Edge.create(v1, v2, edgeProperty, "e1");
+    Edge e2 = Edge.create(v1, v2, edgeProperty, "e1");
+    assertEquals(e1, e2);
+    set.add(e1);
+    assertTrue(set.contains(e2));
+  }
+
+  @Test(timeout = 5000)
+  public void testHashAndEqualsDifferentName() {
+    // edges with different name but everything else same are not equal and have different hash
+    Edge e1 = Edge.create(v1, v2, edgeProperty, "e1");
+    Edge e2 = Edge.create(v1, v2, edgeProperty, "e2");
+    assertNotEquals(e1, e2);
+    set.add(e1);
+    assertFalse(set.contains(e2));
+  }
+}