flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [38/50] [abbrv] flink git commit: [FLINK-1201] [gelly] Renamed tests from TestXyz to XyzITCase
Date Wed, 11 Feb 2015 10:49:40 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestJoinWithVertices.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestJoinWithVertices.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestJoinWithVertices.java
deleted file mode 100644
index 8b0db35..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestJoinWithVertices.java
+++ /dev/null
@@ -1,218 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.Vertex;
-import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
-import org.apache.flink.graph.utils.VertexToTuple2Map;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestJoinWithVertices extends MultipleProgramsTestBase {
-
-	public TestJoinWithVertices(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testJoinWithVertexSet() throws Exception {
-		/*
-		 * Test joinWithVertices with the input DataSet parameter identical
-		 * to the vertex DataSet
-		 */
-        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-        Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-                TestGraphUtils.getLongLongEdgeData(env), env);
-
-        Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices()
-                        .map(new VertexToTuple2Map<Long, Long>()), new AddValuesMapper());
-
-        result.getVertices().writeAsCsv(resultPath);
-        env.execute();
-
-       expectedResult = "1,2\n" +
-	                "2,4\n" +
-	                "3,6\n" +
-	                "4,8\n" +
-	                "5,10\n";
-    }
-
-	@Test
-	public void testWithLessElements() throws Exception {
-	/*
-	 * Test joinWithVertices with the input DataSet passed as a parameter containing
-	 * less elements than the vertex DataSet, but of the same type
-	 */
-        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-        Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-                TestGraphUtils.getLongLongEdgeData(env), env);
-
-        Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices().first(3)
-                        .map(new VertexToTuple2Map<Long, Long>()), new AddValuesMapper());
-
-        result.getVertices().writeAsCsv(resultPath);
-        env.execute();
-
-        expectedResult = "1,2\n" +
-	                "2,4\n" +
-	                "3,6\n" +
-	                "4,4\n" +
-	                "5,5\n";
-    }
-
-	@Test
-	public void testWithDifferentType() throws Exception {
-	/*
-	 * Test joinWithVertices with the input DataSet passed as a parameter containing
-	 * less elements than the vertex DataSet and of a different type(Boolean)
-	 */
-        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-        Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-                TestGraphUtils.getLongLongEdgeData(env), env);
-
-        Graph<Long, Long, Long> result = graph.joinWithVertices(graph.getVertices().first(3)
-                        .map(new ProjectIdWithTrue()), new DoubleIfTrueMapper());
-
-        result.getVertices().writeAsCsv(resultPath);
-        env.execute();
-
-        expectedResult = "1,2\n" +
-	                "2,4\n" +
-	                "3,6\n" +
-	                "4,4\n" +
-	                "5,5\n";
-    }
-
-	@Test
-	public void testWithDifferentKeys() throws Exception {
-		/*
-		 * Test joinWithVertices with an input DataSet containing different keys than the vertex DataSet
-		 * - the iterator becomes empty.
-		 */
-        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-        Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-                TestGraphUtils.getLongLongEdgeData(env), env);
-
-        Graph<Long, Long, Long> result = graph.joinWithVertices(TestGraphUtils.getLongLongTuple2Data(env),
-                new ProjectSecondMapper());
-
-        result.getVertices().writeAsCsv(resultPath);
-        env.execute();
-
-        expectedResult = "1,10\n" +
-	                "2,20\n" +
-	                "3,30\n" +
-	                "4,40\n" +
-	                "5,5\n";
-    }
-
-	@Test
-	public void testWithCustomType() throws Exception {
-		/*
-		 * Test joinWithVertices with a DataSet containing custom parametrised type input values
-		 */
-        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
-        Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-                TestGraphUtils.getLongLongEdgeData(env), env);
-
-        Graph<Long, Long, Long> result = graph.joinWithVertices(TestGraphUtils.getLongCustomTuple2Data(env),
-                new CustomValueMapper());
-
-        result.getVertices().writeAsCsv(resultPath);
-        env.execute();
-
-        expectedResult = "1,10\n" +
-	                "2,20\n" +
-	                "3,30\n" +
-	                "4,40\n" +
-	                "5,5\n";
-    }
-
-	@SuppressWarnings("serial")
-	private static final class AddValuesMapper implements MapFunction<Tuple2<Long, Long>, Long> {
-		public Long map(Tuple2<Long, Long> tuple) throws Exception {
-			return tuple.f0 + tuple.f1;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ProjectIdWithTrue implements MapFunction<Vertex<Long, Long>, Tuple2<Long, Boolean>> {
-        public Tuple2<Long, Boolean> map(Vertex<Long, Long> vertex) throws Exception {
-            return new Tuple2<Long, Boolean>(vertex.getId(), true);
-        }
-    }
-
-	@SuppressWarnings("serial")
-	private static final class DoubleIfTrueMapper implements MapFunction<Tuple2<Long, Boolean>, Long> {
-        public Long map(Tuple2<Long, Boolean> tuple) throws Exception {
-            if(tuple.f1) {
-                return tuple.f0 * 2;
-            }
-            else {
-                return tuple.f0;
-            }
-        }
-    }
-
-	@SuppressWarnings("serial")
-	private static final class ProjectSecondMapper implements MapFunction<Tuple2<Long, Long>, Long> {
-        public Long map(Tuple2<Long, Long> tuple) throws Exception {
-            return tuple.f1;
-        }
-    }
-
-	@SuppressWarnings("serial")
-	private static final class CustomValueMapper implements MapFunction<Tuple2<Long, DummyCustomParameterizedType<Float>>, Long> {
-        public Long map(Tuple2<Long, DummyCustomParameterizedType<Float>> tuple) throws Exception {
-            return (long) tuple.f1.getIntField();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapEdges.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapEdges.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapEdges.java
deleted file mode 100644
index 9eccecc..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapEdges.java
+++ /dev/null
@@ -1,223 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple1;
-import org.apache.flink.graph.Edge;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
-import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestMapEdges extends MultipleProgramsTestBase {
-
-	public TestMapEdges(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testWithSameValue() throws Exception {
-		/*
-		 * Test mapEdges() keeping the same value type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Edge<Long, Long>> mappedEdges = graph.mapEdges(new AddOneMapper()).getEdges();
-		
-		mappedEdges.writeAsCsv(resultPath);
-		env.execute();
-		expectedResult = "1,2,13\n" +
-				"1,3,14\n" +
-				"2,3,24\n" +
-				"3,4,35\n" +
-				"3,5,36\n" + 
-				"4,5,46\n" + 
-				"5,1,52\n";
-	}
-
-	@Test
-	public void testWithStringValue() throws Exception {
-		/*
-		 * Test mapEdges() and change the value type to String
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Edge<Long, String>> mappedEdges = graph.mapEdges(new ToStringMapper()).getEdges();
-		
-		mappedEdges.writeAsCsv(resultPath);
-		env.execute();
-		expectedResult = "1,2,string(12)\n" +
-				"1,3,string(13)\n" +
-				"2,3,string(23)\n" +
-				"3,4,string(34)\n" +
-				"3,5,string(35)\n" + 
-				"4,5,string(45)\n" + 
-				"5,1,string(51)\n";
-	}
-
-	@Test
-	public void testWithTuple1Type() throws Exception {
-		/*
-		 * Test mapEdges() and change the value type to a Tuple1
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Edge<Long, Tuple1<Long>>> mappedEdges = graph.mapEdges(new ToTuple1Mapper()).getEdges();
-		
-		mappedEdges.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,2,(12)\n" +
-				"1,3,(13)\n" +
-				"2,3,(23)\n" +
-				"3,4,(34)\n" +
-				"3,5,(35)\n" + 
-				"4,5,(45)\n" + 
-				"5,1,(51)\n";
-	}
-
-	@Test
-	public void testWithCustomType() throws Exception {
-		/*
-		 * Test mapEdges() and change the value type to a custom type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Edge<Long, DummyCustomType>> mappedEdges = graph.mapEdges(new ToCustomTypeMapper()).getEdges();
-		
-		mappedEdges.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,2,(T,12)\n" +
-			"1,3,(T,13)\n" +
-			"2,3,(T,23)\n" +
-			"3,4,(T,34)\n" +
-			"3,5,(T,35)\n" + 
-			"4,5,(T,45)\n" + 
-			"5,1,(T,51)\n";
-	}
-
-	@Test
-	public void testWithParametrizedCustomType() throws Exception {
-		/*
-		 * Test mapEdges() and change the value type to a parameterized custom type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Edge<Long, DummyCustomParameterizedType<Double>>> mappedEdges = graph.mapEdges(
-				new ToCustomParametrizedTypeMapper()).getEdges();
-		
-		mappedEdges.writeAsCsv(resultPath);
-		env.execute();
-	
-		expectedResult = "1,2,(12.0,12)\n" +
-			"1,3,(13.0,13)\n" +
-			"2,3,(23.0,23)\n" +
-			"3,4,(34.0,34)\n" +
-			"3,5,(35.0,35)\n" + 
-			"4,5,(45.0,45)\n" + 
-			"5,1,(51.0,51)\n";
-	}
-
-	@SuppressWarnings("serial")
-	private static final class AddOneMapper implements MapFunction<Edge<Long, Long>, Long> {
-		public Long map(Edge<Long, Long> edge) throws Exception {
-			return edge.getValue()+1;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToStringMapper implements MapFunction<Edge<Long, Long>, String> {
-		public String map(Edge<Long, Long> edge) throws Exception {
-			return String.format("string(%d)", edge.getValue());
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToTuple1Mapper implements MapFunction<Edge<Long, Long>, Tuple1<Long>> {
-		public Tuple1<Long> map(Edge<Long, Long> edge) throws Exception {
-			Tuple1<Long> tupleValue = new Tuple1<Long>();
-			tupleValue.setFields(edge.getValue());
-			return tupleValue;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToCustomTypeMapper implements MapFunction<Edge<Long, Long>, DummyCustomType> {
-		public DummyCustomType map(Edge<Long, Long> edge) throws Exception {
-			DummyCustomType dummyValue = new DummyCustomType();
-			dummyValue.setIntField(edge.getValue().intValue());						
-			return dummyValue;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToCustomParametrizedTypeMapper implements MapFunction<Edge<Long, Long>, 
-		DummyCustomParameterizedType<Double>> {
-
-		public DummyCustomParameterizedType<Double> map(Edge<Long, Long> edge) throws Exception {
-			DummyCustomParameterizedType<Double> dummyValue = new DummyCustomParameterizedType<Double>();
-			dummyValue.setIntField(edge.getValue().intValue());
-			dummyValue.setTField(new Double(edge.getValue()));						
-			return dummyValue;
-		}
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapVertices.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapVertices.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapVertices.java
deleted file mode 100644
index c4d44b0..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestMapVertices.java
+++ /dev/null
@@ -1,233 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple1;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.Vertex;
-import org.apache.flink.graph.test.TestGraphUtils.DummyCustomParameterizedType;
-import org.apache.flink.graph.test.TestGraphUtils.DummyCustomType;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestMapVertices extends MultipleProgramsTestBase {
-
-	public TestMapVertices(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testWithSameValue() throws Exception {
-		/*
-		 * Test mapVertices() keeping the same value type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Vertex<Long, Long>> mappedVertices = graph.mapVertices(new AddOneMapper()).getVertices();
-		
-		mappedVertices.writeAsCsv(resultPath);
-		env.execute();
-		expectedResult = "1,2\n" +
-			"2,3\n" +
-			"3,4\n" +
-			"4,5\n" +
-			"5,6\n";
-	}
-
-	@Test
-	public void testWithStringValue() throws Exception {
-		/*
-		 * Test mapVertices() and change the value type to String
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Vertex<Long, String>> mappedVertices = graph.mapVertices(new ToStringMapper()).getVertices();
-		
-		mappedVertices.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,one\n" +
-			"2,two\n" +
-			"3,three\n" +
-			"4,four\n" +
-			"5,five\n";
-	}
-
-	@Test
-	public void testWithtuple1Value() throws Exception {
-		/*
-		 * Test mapVertices() and change the value type to a Tuple1
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Vertex<Long, Tuple1<Long>>> mappedVertices = graph.mapVertices(new ToTuple1Mapper()).getVertices();
-		
-		mappedVertices.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,(1)\n" +
-			"2,(2)\n" +
-			"3,(3)\n" +
-			"4,(4)\n" +
-			"5,(5)\n";
-	}
-
-	@Test
-	public void testWithCustomType() throws Exception {
-		/*
-		 * Test mapVertices() and change the value type to a custom type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Vertex<Long, DummyCustomType>> mappedVertices = graph.mapVertices(new ToCustomTypeMapper()).getVertices();
-		
-		mappedVertices.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,(T,1)\n" +
-			"2,(T,2)\n" +
-			"3,(T,3)\n" +
-			"4,(T,4)\n" +
-			"5,(T,5)\n";
-	}
-
-	@Test
-	public void testWithCustomParametrizedType() throws Exception {
-		/*
-		 * Test mapVertices() and change the value type to a parameterized custom type
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		DataSet<Vertex<Long, DummyCustomParameterizedType<Double>>> mappedVertices = graph.mapVertices(
-				new ToCustomParametrizedTypeMapper()).getVertices();
-		
-		mappedVertices.writeAsCsv(resultPath);
-		env.execute();
-	
-		expectedResult = "1,(1.0,1)\n" +
-			"2,(2.0,2)\n" +
-			"3,(3.0,3)\n" +
-			"4,(4.0,4)\n" +
-			"5,(5.0,5)\n";
-	}
-
-	@SuppressWarnings("serial")
-	private static final class AddOneMapper implements MapFunction<Vertex<Long, Long>, Long> {
-		public Long map(Vertex<Long, Long> value) throws Exception {
-			return value.getValue()+1;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToStringMapper implements MapFunction<Vertex<Long, Long>, String> {
-		public String map(Vertex<Long, Long> vertex) throws Exception {
-			String stringValue;
-			if (vertex.getValue() == 1) {
-				stringValue = "one";
-			}
-			else if (vertex.getValue() == 2) {
-				stringValue = "two";
-			}
-			else if (vertex.getValue() == 3) {
-				stringValue = "three";
-			}
-			else if (vertex.getValue() == 4) {
-				stringValue = "four";
-			}
-			else if (vertex.getValue() == 5) {
-				stringValue = "five";
-			}
-			else {
-				stringValue = "";
-			}
-			return stringValue;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToTuple1Mapper implements MapFunction<Vertex<Long, Long>, Tuple1<Long>> {
-		public Tuple1<Long> map(Vertex<Long, Long> vertex) throws Exception {
-			Tuple1<Long> tupleValue = new Tuple1<Long>();
-			tupleValue.setFields(vertex.getValue());
-			return tupleValue;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToCustomTypeMapper implements MapFunction<Vertex<Long, Long>, DummyCustomType> {
-		public DummyCustomType map(Vertex<Long, Long> vertex) throws Exception {
-			DummyCustomType dummyValue = new DummyCustomType();
-			dummyValue.setIntField(vertex.getValue().intValue());						
-			return dummyValue;
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class ToCustomParametrizedTypeMapper implements MapFunction<Vertex<Long, Long>, 
-		DummyCustomParameterizedType<Double>> {
-		
-		public DummyCustomParameterizedType<Double> map(Vertex<Long, Long> vertex) throws Exception {
-			DummyCustomParameterizedType<Double> dummyValue = new DummyCustomParameterizedType<Double>();
-			dummyValue.setIntField(vertex.getValue().intValue());
-			dummyValue.setTField(new Double(vertex.getValue()));						
-			return dummyValue;
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnEdgesMethods.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnEdgesMethods.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnEdgesMethods.java
deleted file mode 100644
index 7a02ffe..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnEdgesMethods.java
+++ /dev/null
@@ -1,317 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.graph.Edge;
-import org.apache.flink.graph.EdgeDirection;
-import org.apache.flink.graph.EdgesFunction;
-import org.apache.flink.graph.EdgesFunctionWithVertexValue;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.Vertex;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestReduceOnEdgesMethods extends MultipleProgramsTestBase {
-
-	public TestReduceOnEdgesMethods(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testLowestWeightOutNeighbor() throws Exception {
-		/*
-		 * Get the lowest-weight out-neighbor
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor = 
-				graph.reduceOnEdges(new SelectMinWeightNeighbor(), EdgeDirection.OUT);
-		verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
-		env.execute();
-	
-		expectedResult = "1,2\n" +
-				"2,3\n" + 
-				"3,4\n" +
-				"4,5\n" + 
-				"5,1\n";
-	}
-
-	@Test
-	public void testLowestWeightInNeighbor() throws Exception {
-		/*
-		 * Get the lowest-weight in-neighbor
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor = 
-				graph.reduceOnEdges(new SelectMinWeightInNeighbor(), EdgeDirection.IN);
-		verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,5\n" +
-					"2,1\n" + 
-					"3,1\n" +
-					"4,3\n" + 
-					"5,3\n";
-	}
-
-	@Test
-	public void testMaxWeightEdge() throws Exception {
-		/*
-		 * Get the maximum weight among all edges
-		 * of a vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithMaxEdgeWeight = 
-				graph.reduceOnEdges(new SelectMaxWeightNeighbor(), EdgeDirection.ALL);
-		verticesWithMaxEdgeWeight.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,51\n" +
-				"2,23\n" + 
-				"3,35\n" +
-				"4,45\n" + 
-				"5,51\n";
-	}
-
-	@Test
-	public void testLowestWeightOutNeighborNoValue() throws Exception {
-		/*
-		 * Get the lowest-weight out-neighbor
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor = 
-				graph.reduceOnEdges(new SelectMinWeightNeighborNoValue(), EdgeDirection.OUT);
-		verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,2\n" +
-				"2,3\n" + 
-				"3,4\n" +
-				"4,5\n" + 
-				"5,1\n";
-	}
-
-	@Test
-	public void testLowestWeightInNeighborNoValue() throws Exception {
-		/*
-		 * Get the lowest-weight in-neighbor
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithLowestOutNeighbor = 
-				graph.reduceOnEdges(new SelectMinWeightInNeighborNoValue(), EdgeDirection.IN);
-		verticesWithLowestOutNeighbor.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,5\n" +
-				"2,1\n" + 
-				"3,1\n" +
-				"4,3\n" + 
-				"5,3\n";
-	}
-
-	@Test
-	public void testMaxWeightAllNeighbors() throws Exception {
-		/*
-		 * Get the maximum weight among all edges
-		 * of a vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithMaxEdgeWeight = 
-				graph.reduceOnEdges(new SelectMaxWeightNeighborNoValue(), EdgeDirection.ALL);
-		verticesWithMaxEdgeWeight.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,51\n" +
-				"2,23\n" + 
-				"3,35\n" +
-				"4,45\n" + 
-				"5,51\n";
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMinWeightNeighbor implements EdgesFunctionWithVertexValue<Long, Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(
-				Vertex<Long, Long> v,
-				Iterable<Edge<Long, Long>> edges) {
-			
-			long weight = Long.MAX_VALUE;
-			long minNeighorId = 0;
-			
-			for (Edge<Long, Long> edge: edges) {
-				if (edge.getValue() < weight) {
-					weight = edge.getValue();
-					minNeighorId = edge.getTarget();
-				}
-			}
-			return new Tuple2<Long, Long>(v.getId(), minNeighorId);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMaxWeightNeighbor implements EdgesFunctionWithVertexValue<Long, Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(Vertex<Long, Long> v,
-				Iterable<Edge<Long, Long>> edges) {
-			
-			long weight = Long.MIN_VALUE;
-
-			for (Edge<Long, Long> edge: edges) {
-				if (edge.getValue() > weight) {
-					weight = edge.getValue();
-				}
-			}
-			return new Tuple2<Long, Long>(v.getId(), weight);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMinWeightNeighborNoValue implements EdgesFunction<Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(Iterable<Tuple2<Long, Edge<Long, Long>>> edges) {
-
-			long weight = Long.MAX_VALUE;
-			long minNeighorId = 0;
-			long vertexId = -1;
-			long i=0;
-
-			for (Tuple2<Long, Edge<Long, Long>> edge: edges) {
-				if (edge.f1.getValue() < weight) {
-					weight = edge.f1.getValue();
-					minNeighorId = edge.f1.getTarget();
-				}
-				if (i==0) {
-					vertexId = edge.f0;
-				} i++;
-			}
-			return new Tuple2<Long, Long>(vertexId, minNeighorId);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMaxWeightNeighborNoValue implements EdgesFunction<Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(Iterable<Tuple2<Long, Edge<Long, Long>>> edges) {
-			
-			long weight = Long.MIN_VALUE;
-			long vertexId = -1;
-			long i=0;
-
-			for (Tuple2<Long, Edge<Long, Long>> edge: edges) {
-				if (edge.f1.getValue() > weight) {
-					weight = edge.f1.getValue();
-				}
-				if (i==0) {
-					vertexId = edge.f0;
-				} i++;
-			}
-			return new Tuple2<Long, Long>(vertexId, weight);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMinWeightInNeighbor implements EdgesFunctionWithVertexValue<Long, Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(
-				Vertex<Long, Long> v,
-				Iterable<Edge<Long, Long>> edges) {
-			
-			long weight = Long.MAX_VALUE;
-			long minNeighorId = 0;
-			
-			for (Edge<Long, Long> edge: edges) {
-				if (edge.getValue() < weight) {
-					weight = edge.getValue();
-					minNeighorId = edge.getSource();
-				}
-			}
-			return new Tuple2<Long, Long>(v.getId(), minNeighorId);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SelectMinWeightInNeighborNoValue implements EdgesFunction<Long, Long, Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateEdges(Iterable<Tuple2<Long, Edge<Long, Long>>> edges) {
-			
-			long weight = Long.MAX_VALUE;
-			long minNeighorId = 0;
-			long vertexId = -1;
-			long i=0;
-
-			for (Tuple2<Long, Edge<Long, Long>> edge: edges) {
-				if (edge.f1.getValue() < weight) {
-					weight = edge.f1.getValue();
-					minNeighorId = edge.f1.getSource();
-				}
-				if (i==0) {
-					vertexId = edge.f0;
-				} i++;
-			}
-			return new Tuple2<Long, Long>(vertexId, minNeighorId);
-		}
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnNeighborMethods.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnNeighborMethods.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnNeighborMethods.java
deleted file mode 100644
index e64eacf..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestReduceOnNeighborMethods.java
+++ /dev/null
@@ -1,303 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import java.util.Iterator;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.graph.Edge;
-import org.apache.flink.graph.EdgeDirection;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.NeighborsFunction;
-import org.apache.flink.graph.NeighborsFunctionWithVertexValue;
-import org.apache.flink.graph.Vertex;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestReduceOnNeighborMethods extends MultipleProgramsTestBase {
-
-	public TestReduceOnNeighborMethods(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testSumOfOutNeighbors() throws Exception {
-		/*
-		 * Get the sum of out-neighbor values
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues = 
-				graph.reduceOnNeighbors(new SumOutNeighbors(), EdgeDirection.OUT);
-
-		verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
-		env.execute();
-		expectedResult = "1,5\n" +
-				"2,3\n" + 
-				"3,9\n" +
-				"4,5\n" + 
-				"5,1\n";
-	}
-
-	@Test
-	public void testSumOfInNeighbors() throws Exception {
-		/*
-		 * Get the sum of in-neighbor values
-		 * times the edge weights for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSum = 
-				graph.reduceOnNeighbors(new SumInNeighbors(), EdgeDirection.IN);		
-
-		verticesWithSum.writeAsCsv(resultPath);
-		env.execute();
-		expectedResult = "1,255\n" +
-				"2,12\n" + 
-				"3,59\n" +
-				"4,102\n" + 
-				"5,285\n";
-	}
-
-	@Test
-	public void testSumOfOAllNeighbors() throws Exception {
-		/*
-		 * Get the sum of all neighbor values
-		 * including own vertex value
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues = 
-				graph.reduceOnNeighbors(new SumAllNeighbors(), EdgeDirection.ALL);
-
-		verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,11\n" +
-				"2,6\n" + 
-				"3,15\n" +
-				"4,12\n" + 
-				"5,13\n";
-	}
-
-	@Test
-	public void testSumOfOutNeighborsNoValue() throws Exception {
-		/*
-		 * Get the sum of out-neighbor values
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues = 
-				graph.reduceOnNeighbors(new SumOutNeighborsNoValue(), EdgeDirection.OUT);
-
-		verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
-		env.execute();
-
-		expectedResult = "1,5\n" +
-				"2,3\n" + 
-				"3,9\n" +
-				"4,5\n" + 
-				"5,1\n";
-	}
-
-	@Test
-	public void testSumOfInNeighborsNoValue() throws Exception {
-		/*
-		 * Get the sum of in-neighbor values
-		 * times the edge weights for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSum = 
-				graph.reduceOnNeighbors(new SumInNeighborsNoValue(), EdgeDirection.IN);
-
-		verticesWithSum.writeAsCsv(resultPath);
-		env.execute();
-	
-		expectedResult = "1,255\n" +
-				"2,12\n" + 
-				"3,59\n" +
-				"4,102\n" + 
-				"5,285\n";
-	}
-
-	@Test
-	public void testSumOfAllNeighborsNoValue() throws Exception {
-		/*
-		 * Get the sum of all neighbor values
-		 * for each vertex
-         */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env), 
-				TestGraphUtils.getLongLongEdgeData(env), env);
-
-		DataSet<Tuple2<Long, Long>> verticesWithSumOfOutNeighborValues = 
-				graph.reduceOnNeighbors(new SumAllNeighborsNoValue(), EdgeDirection.ALL);
-
-		verticesWithSumOfOutNeighborValues.writeAsCsv(resultPath);
-		env.execute();
-	
-		expectedResult = "1,10\n" +
-				"2,4\n" + 
-				"3,12\n" +
-				"4,8\n" + 
-				"5,8\n";
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumOutNeighbors implements NeighborsFunctionWithVertexValue<Long, Long, Long, 
-	Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateNeighbors(Vertex<Long, Long> vertex,
-				Iterable<Tuple2<Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-			
-			long sum = 0;
-			for (Tuple2<Edge<Long, Long>, Vertex<Long, Long>> neighbor : neighbors) {
-				sum += neighbor.f1.getValue();
-			}
-			return new Tuple2<Long, Long>(vertex.getId(), sum);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumInNeighbors implements NeighborsFunctionWithVertexValue<Long, Long, Long, 
-		Tuple2<Long, Long>> {
-		
-		public Tuple2<Long, Long> iterateNeighbors(Vertex<Long, Long> vertex,
-				Iterable<Tuple2<Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-		
-			long sum = 0;
-			for (Tuple2<Edge<Long, Long>, Vertex<Long, Long>> neighbor : neighbors) {
-				sum += neighbor.f0.getValue() * neighbor.f1.getValue();
-			}
-			return new Tuple2<Long, Long>(vertex.getId(), sum);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumAllNeighbors implements NeighborsFunctionWithVertexValue<Long, Long, Long, 
-		Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateNeighbors(Vertex<Long, Long> vertex,
-		Iterable<Tuple2<Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-	
-			long sum = 0;
-			for (Tuple2<Edge<Long, Long>, Vertex<Long, Long>> neighbor : neighbors) {
-				sum += neighbor.f1.getValue();
-			}
-			return new Tuple2<Long, Long>(vertex.getId(), sum + vertex.getValue());
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumOutNeighborsNoValue implements NeighborsFunction<Long, Long, Long, 
-		Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateNeighbors(
-				Iterable<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-
-			long sum = 0;
-			Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>> next = null;
-			Iterator<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighborsIterator =
-					neighbors.iterator();
-			while(neighborsIterator.hasNext()) {
-				next = neighborsIterator.next();
-				sum += next.f2.getValue();
-			}
-			return new Tuple2<Long, Long>(next.f0, sum);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumInNeighborsNoValue implements NeighborsFunction<Long, Long, Long, 
-		Tuple2<Long, Long>> {
-		
-		public Tuple2<Long, Long> iterateNeighbors(
-				Iterable<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-		
-			long sum = 0;
-			Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>> next = null;
-			Iterator<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighborsIterator =
-					neighbors.iterator();
-			while(neighborsIterator.hasNext()) {
-				next = neighborsIterator.next();
-				sum += next.f2.getValue() * next.f1.getValue();
-			}
-			return new Tuple2<Long, Long>(next.f0, sum);
-		}
-	}
-
-	@SuppressWarnings("serial")
-	private static final class SumAllNeighborsNoValue implements NeighborsFunction<Long, Long, Long, 
-		Tuple2<Long, Long>> {
-
-		public Tuple2<Long, Long> iterateNeighbors(
-				Iterable<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighbors) {
-	
-			long sum = 0;
-			Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>> next = null;
-			Iterator<Tuple3<Long, Edge<Long, Long>, Vertex<Long, Long>>> neighborsIterator =
-					neighbors.iterator();
-			while(neighborsIterator.hasNext()) {
-				next = neighborsIterator.next();
-				sum += next.f2.getValue();
-			}
-			return new Tuple2<Long, Long>(next.f0, sum);
-		}
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestVertexCentricConnectedComponents.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestVertexCentricConnectedComponents.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestVertexCentricConnectedComponents.java
deleted file mode 100644
index 614ddd5..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestVertexCentricConnectedComponents.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.flink.graph.test;
-
-import java.io.BufferedReader;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.graph.Edge;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.graph.Vertex;
-import org.apache.flink.graph.spargel.MessageIterator;
-import org.apache.flink.graph.spargel.MessagingFunction;
-import org.apache.flink.graph.spargel.VertexUpdateFunction;
-import org.apache.flink.test.testdata.ConnectedComponentsData;
-import org.apache.flink.test.util.JavaProgramTestBase;
-import org.apache.flink.types.NullValue;
-
-@SuppressWarnings("serial")
-public class TestVertexCentricConnectedComponents extends JavaProgramTestBase {
-
-	private static final long SEED = 9487520347802987L;
-	
-	private static final int NUM_VERTICES = 1000;
-	
-	private static final int NUM_EDGES = 10000;
-
-	private String resultPath;
-	
-	
-	@Override
-	protected void preSubmit() throws Exception {
-		resultPath = getTempFilePath("results");
-	}
-	
-	@Override
-	protected void testProgram() throws Exception {
-		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		DataSet<Long> vertexIds = env.generateSequence(1, NUM_VERTICES);
-		DataSet<String> edgeString = env.fromElements(ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED).split("\n"));
-		
-		DataSet<Edge<Long, NullValue>> edges = edgeString.map(new EdgeParser());
-		
-		DataSet<Vertex<Long, Long>> initialVertices = vertexIds.map(new IdAssigner());
-		Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env); 
-		
-		Graph<Long, Long, NullValue> result = graph.runVertexCentricIteration(new CCUpdater(), new CCMessager(), 100);
-		
-		result.getVertices().writeAsCsv(resultPath, "\n", " ");
-		env.execute();
-	}
-	
-	public static final class CCUpdater extends VertexUpdateFunction<Long, Long, Long> {
-		@Override
-		public void updateVertex(Long vertexKey, Long vertexValue, MessageIterator<Long> inMessages) {
-			long min = Long.MAX_VALUE;
-			for (long msg : inMessages) {
-				min = Math.min(min, msg);
-			}
-			if (min < vertexValue) {
-				setNewVertexValue(min);
-			}
-		}
-	}
-	
-	public static final class CCMessager extends MessagingFunction<Long, Long, Long, NullValue> {
-		@Override
-		public void sendMessages(Long vertexId, Long componentId) {
-			sendMessageToAllNeighbors(componentId);
-		}
-	}
-	
-	/**
-	 * A map function that takes a Long value and creates a 2-tuple out of it:
-	 * <pre>(Long value) -> (value, value)</pre>
-	 */
-	public static final class IdAssigner implements MapFunction<Long, Vertex<Long, Long>> {
-		@Override
-		public Vertex<Long, Long> map(Long value) {
-			return new Vertex<Long, Long>(value, value);
-		}
-	}
-
-	@Override
-	protected void postSubmit() throws Exception {
-		for (BufferedReader reader : getResultReader(resultPath)) {
-			ConnectedComponentsData.checkOddEvenResult(reader);
-		}
-	}
-	
-	public static final class EdgeParser extends RichMapFunction<String, Edge<Long, NullValue>> {
-		public Edge<Long, NullValue> map(String value) {
-			String[] nums = value.split(" ");
-			return new Edge<Long, NullValue>(Long.parseLong(nums[0]), Long.parseLong(nums[1]), 
-					NullValue.getInstance());
-		}
-	}
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestWeaklyConnected.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestWeaklyConnected.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestWeaklyConnected.java
deleted file mode 100644
index f5b6d9d..0000000
--- a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/TestWeaklyConnected.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.graph.test;
-
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.graph.Graph;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-
-@RunWith(Parameterized.class)
-public class TestWeaklyConnected extends MultipleProgramsTestBase {
-
-	public TestWeaklyConnected(MultipleProgramsTestBase.ExecutionMode mode){
-		super(mode);
-	}
-
-    private String resultPath;
-    private String expectedResult;
-
-    @Rule
-	public TemporaryFolder tempFolder = new TemporaryFolder();
-
-	@Before
-	public void before() throws Exception{
-		resultPath = tempFolder.newFile().toURI().toString();
-	}
-
-	@After
-	public void after() throws Exception{
-		compareResultsByLinesInMemory(expectedResult, resultPath);
-	}
-
-	@Test
-	public void testWithConnectedDirected() throws Exception {
-		/*
-		 * Test isWeaklyConnected() with a connected, directed graph
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env);
-		
-		graph.isWeaklyConnected(10).writeAsText(resultPath);
-		
-		env.execute();
-		expectedResult = "true\n";
-	}
-
-	@Test
-	public void testWithDisconnectedDirected() throws Exception {
-		/*
-		 * Test isWeaklyConnected() with a disconnected, directed graph
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getDisconnectedLongLongEdgeData(env), env);
-		
-		graph.isWeaklyConnected(10).writeAsText(resultPath);
-		
-		env.execute();
-		expectedResult = "false\n";
-	}
-
-	@Test
-	public void testWithConnectedUndirected() throws Exception {
-		/*
-		 * Test isWeaklyConnected() with a connected, undirected graph
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getLongLongEdgeData(env), env).getUndirected();
-		
-		graph.isWeaklyConnected(10).writeAsText(resultPath);
-		
-		env.execute();
-		expectedResult = "true\n";
-	}
-
-	@Test
-	public void testWithDisconnectedUndirected() throws Exception {
-		/*
-		 * Test isWeaklyConnected() with a disconnected, undirected graph
-		 */
-		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
-				TestGraphUtils.getDisconnectedLongLongEdgeData(env), env).getUndirected();
-		
-		graph.isWeaklyConnected(10).writeAsText(resultPath);
-		
-		env.execute();
-		expectedResult = "false\n";
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java
new file mode 100644
index 0000000..380e027
--- /dev/null
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/VertexCentricConnectedComponentsITCase.java
@@ -0,0 +1,118 @@
+/*
+ * 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.flink.graph.test;
+
+import java.io.BufferedReader;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.graph.Edge;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.graph.Vertex;
+import org.apache.flink.graph.spargel.MessageIterator;
+import org.apache.flink.graph.spargel.MessagingFunction;
+import org.apache.flink.graph.spargel.VertexUpdateFunction;
+import org.apache.flink.test.testdata.ConnectedComponentsData;
+import org.apache.flink.test.util.JavaProgramTestBase;
+import org.apache.flink.types.NullValue;
+
+@SuppressWarnings("serial")
+public class VertexCentricConnectedComponentsITCase extends JavaProgramTestBase {
+
+	private static final long SEED = 9487520347802987L;
+	
+	private static final int NUM_VERTICES = 1000;
+	
+	private static final int NUM_EDGES = 10000;
+
+	private String resultPath;
+	
+	
+	@Override
+	protected void preSubmit() throws Exception {
+		resultPath = getTempFilePath("results");
+	}
+	
+	@Override
+	protected void testProgram() throws Exception {
+		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		DataSet<Long> vertexIds = env.generateSequence(1, NUM_VERTICES);
+		DataSet<String> edgeString = env.fromElements(ConnectedComponentsData.getRandomOddEvenEdges(NUM_EDGES, NUM_VERTICES, SEED).split("\n"));
+		
+		DataSet<Edge<Long, NullValue>> edges = edgeString.map(new EdgeParser());
+		
+		DataSet<Vertex<Long, Long>> initialVertices = vertexIds.map(new IdAssigner());
+		Graph<Long, Long, NullValue> graph = Graph.fromDataSet(initialVertices, edges, env); 
+		
+		Graph<Long, Long, NullValue> result = graph.runVertexCentricIteration(new CCUpdater(), new CCMessager(), 100);
+		
+		result.getVertices().writeAsCsv(resultPath, "\n", " ");
+		env.execute();
+	}
+	
+	public static final class CCUpdater extends VertexUpdateFunction<Long, Long, Long> {
+		@Override
+		public void updateVertex(Long vertexKey, Long vertexValue, MessageIterator<Long> inMessages) {
+			long min = Long.MAX_VALUE;
+			for (long msg : inMessages) {
+				min = Math.min(min, msg);
+			}
+			if (min < vertexValue) {
+				setNewVertexValue(min);
+			}
+		}
+	}
+	
+	public static final class CCMessager extends MessagingFunction<Long, Long, Long, NullValue> {
+		@Override
+		public void sendMessages(Long vertexId, Long componentId) {
+			sendMessageToAllNeighbors(componentId);
+		}
+	}
+	
+	/**
+	 * A map function that takes a Long value and creates a 2-tuple out of it:
+	 * <pre>(Long value) -> (value, value)</pre>
+	 */
+	public static final class IdAssigner implements MapFunction<Long, Vertex<Long, Long>> {
+		@Override
+		public Vertex<Long, Long> map(Long value) {
+			return new Vertex<Long, Long>(value, value);
+		}
+	}
+
+	@Override
+	protected void postSubmit() throws Exception {
+		for (BufferedReader reader : getResultReader(resultPath)) {
+			ConnectedComponentsData.checkOddEvenResult(reader);
+		}
+	}
+	
+	public static final class EdgeParser extends RichMapFunction<String, Edge<Long, NullValue>> {
+		public Edge<Long, NullValue> map(String value) {
+			String[] nums = value.split(" ");
+			return new Edge<Long, NullValue>(Long.parseLong(nums[0]), Long.parseLong(nums[1]), 
+					NullValue.getInstance());
+		}
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/6296c394/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/WeaklyConnectedITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/WeaklyConnectedITCase.java b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/WeaklyConnectedITCase.java
new file mode 100644
index 0000000..dd08f47
--- /dev/null
+++ b/flink-staging/flink-gelly/src/test/java/org/apache/flink/graph/test/WeaklyConnectedITCase.java
@@ -0,0 +1,118 @@
+/*
+ * 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.flink.graph.test;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.graph.Graph;
+import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class WeaklyConnectedITCase extends MultipleProgramsTestBase {
+
+	public WeaklyConnectedITCase(MultipleProgramsTestBase.ExecutionMode mode){
+		super(mode);
+	}
+
+    private String resultPath;
+    private String expectedResult;
+
+    @Rule
+	public TemporaryFolder tempFolder = new TemporaryFolder();
+
+	@Before
+	public void before() throws Exception{
+		resultPath = tempFolder.newFile().toURI().toString();
+	}
+
+	@After
+	public void after() throws Exception{
+		compareResultsByLinesInMemory(expectedResult, resultPath);
+	}
+
+	@Test
+	public void testWithConnectedDirected() throws Exception {
+		/*
+		 * Test isWeaklyConnected() with a connected, directed graph
+		 */
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+				TestGraphUtils.getLongLongEdgeData(env), env);
+		
+		graph.isWeaklyConnected(10).writeAsText(resultPath);
+		
+		env.execute();
+		expectedResult = "true\n";
+	}
+
+	@Test
+	public void testWithDisconnectedDirected() throws Exception {
+		/*
+		 * Test isWeaklyConnected() with a disconnected, directed graph
+		 */
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+				TestGraphUtils.getDisconnectedLongLongEdgeData(env), env);
+		
+		graph.isWeaklyConnected(10).writeAsText(resultPath);
+		
+		env.execute();
+		expectedResult = "false\n";
+	}
+
+	@Test
+	public void testWithConnectedUndirected() throws Exception {
+		/*
+		 * Test isWeaklyConnected() with a connected, undirected graph
+		 */
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+				TestGraphUtils.getLongLongEdgeData(env), env).getUndirected();
+		
+		graph.isWeaklyConnected(10).writeAsText(resultPath);
+		
+		env.execute();
+		expectedResult = "true\n";
+	}
+
+	@Test
+	public void testWithDisconnectedUndirected() throws Exception {
+		/*
+		 * Test isWeaklyConnected() with a disconnected, undirected graph
+		 */
+		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		
+		Graph<Long, Long, Long> graph = Graph.fromDataSet(TestGraphUtils.getLongLongVertexData(env),
+				TestGraphUtils.getDisconnectedLongLongEdgeData(env), env).getUndirected();
+		
+		graph.isWeaklyConnected(10).writeAsText(resultPath);
+		
+		env.execute();
+		expectedResult = "false\n";
+	}
+}


Mime
View raw message