flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From trohrm...@apache.org
Subject [3/4] flink git commit: [FLINK-3013] [gelly] Incorrect package declaration in GellyScalaAPICompletenessTest.scala
Date Wed, 18 Nov 2015 15:05:59 GMT
[FLINK-3013] [gelly] Incorrect package declaration in GellyScalaAPICompletenessTest.scala

This closes #1356.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ee648925
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ee648925
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ee648925

Branch: refs/heads/master
Commit: ee64892591de67a60023fd202704370cd18d1b26
Parents: 5d37fe1
Author: smarthi <smarthi@apache.org>
Authored: Sun Nov 15 14:44:14 2015 -0500
Committer: Till Rohrmann <trohrmann@apache.org>
Committed: Wed Nov 18 15:51:49 2015 +0100

----------------------------------------------------------------------
 .../org/apache/flink/graph/scala/Graph.scala    |  12 +-
 .../scala/example/ConnectedComponents.scala     |  11 +-
 .../example/GSASingleSourceShortestPaths.scala  |  20 +-
 .../graph/scala/example/GraphMetrics.scala      |   8 +-
 .../example/SingleSourceShortestPaths.scala     |  12 +-
 .../test/GellyScalaAPICompletenessTest.scala    |   2 +-
 .../flink/graph/scala/test/TestGraphUtils.scala |   4 +-
 .../scala/test/operations/DegreesITCase.scala   |  16 +-
 .../operations/GraphCreationWithCsvITCase.scala |  65 ++---
 .../test/operations/GraphMutationsITCase.scala  |  38 +--
 .../test/operations/GraphOperationsITCase.scala |  65 ++---
 .../test/operations/JoinWithEdgesITCase.scala   |   7 +-
 .../operations/JoinWithVerticesITCase.scala     |   7 +-
 .../scala/test/operations/MapEdgesITCase.scala  |   4 +-
 .../test/operations/MapVerticesITCase.scala     |   8 +-
 .../operations/ReduceOnEdgesMethodsITCase.scala |   4 +-
 .../ReduceOnNeighborMethodsITCase.scala         |   4 +-
 .../flink/graph/example/IncrementalSSSP.java    |   6 +-
 .../apache/flink/graph/gsa/GSACompilerTest.java |   6 +-
 .../flink/graph/gsa/GSATranslationTest.java     |   8 +-
 .../graph/spargel/SpargelCompilerTest.java      |  12 +-
 .../graph/spargel/SpargelTranslationTest.java   |  20 +-
 .../test/CollectionModeSuperstepITCase.java     |   2 +-
 .../test/GatherSumApplyConfigurationITCase.java |   5 +-
 .../flink/graph/test/GatherSumApplyITCase.java  |   2 +-
 .../apache/flink/graph/test/TestGraphUtils.java | 291 +++++++++----------
 .../test/VertexCentricConfigurationITCase.java  |  38 ++-
 .../test/example/IncrementalSSSPITCase.java     |   2 +-
 .../graph/test/example/MusicProfilesITCase.java |   2 +-
 ...ctedComponentsWithRandomisedEdgesITCase.java |   4 +-
 .../graph/test/library/PageRankITCase.java      |  16 +-
 .../graph/test/library/TriangleCountITCase.java |   4 +-
 .../test/operations/GraphCreationITCase.java    |   7 +-
 .../GraphCreationWithMapperITCase.java          |   2 +-
 .../test/operations/GraphMutationsITCase.java   | 222 +++++++-------
 .../test/operations/GraphOperationsITCase.java  |  10 +-
 .../test/operations/JoinWithEdgesITCase.java    |  11 +-
 .../test/operations/JoinWithVerticesITCase.java |   2 +-
 .../graph/test/operations/MapEdgesITCase.java   |   4 +-
 .../test/operations/MapVerticesITCase.java      |   4 +-
 .../operations/ReduceOnEdgesMethodsITCase.java  |  35 +--
 .../ReduceOnEdgesWithExceptionITCase.java       |   4 +-
 .../ReduceOnNeighborMethodsITCase.java          |  62 ++--
 .../ReduceOnNeighborsWithExceptionITCase.java   |   2 +-
 .../apache/flink/test/util/TestBaseUtils.java   |  30 +-
 45 files changed, 518 insertions(+), 582 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/Graph.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/Graph.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/Graph.scala
index 3a0843a..fe04887 100644
--- a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/Graph.scala
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/Graph.scala
@@ -170,7 +170,7 @@ object Graph {
   /**
   * Creates a Graph with from a CSV file of vertices and a CSV file of edges
   * 
-  * @param The Execution Environment.
+  * @param env Execution Environment.
   * @param pathEdges The file path containing the edges.
   * @param readVertices Defines whether the vertices have associated values.
   * If set to false, the vertex input is ignored and vertices are created from the edges file.
@@ -868,7 +868,7 @@ TypeInformation : ClassTag](jgraph: jg.Graph[K, VV, EV]) {
   * Adds the list of vertices, passed as input, to the graph.
   * If the vertices already exist in the graph, they will not be added once more.
   *
-  * @param verticesToAdd the list of vertices to add
+  * @param vertices the list of vertices to add
   * @return the new graph containing the existing and newly added vertices
   */
   def addVertices(vertices: List[Vertex[K, VV]]): Graph[K, VV, EV] = {
@@ -881,7 +881,7 @@ TypeInformation : ClassTag](jgraph: jg.Graph[K, VV, EV]) {
   * When adding an edge for a non-existing set of vertices,
   * the edge is considered invalid and ignored.
   *
-  * @param newEdges the data set of edges to be added
+  * @param edges the data set of edges to be added
   * @return a new graph containing the existing edges plus the newly added edges.
   */
   def addEdges(edges: List[Edge[K, EV]]): Graph[K, VV, EV] = {
@@ -916,7 +916,7 @@ TypeInformation : ClassTag](jgraph: jg.Graph[K, VV, EV]) {
     /**
    * Removes the given vertex and its edges from the graph.
    *
-   * @param vertex the vertex to remove
+   * @param vertices list of vertices to remove
    * @return the new graph containing the existing vertices and edges without
    *         the removed vertex and its edges
    */
@@ -938,7 +938,7 @@ TypeInformation : ClassTag](jgraph: jg.Graph[K, VV, EV]) {
   /**
    * Removes all the edges that match the edges in the given data set from the graph.
    *
-   * @param edgesToBeRemoved the list of edges to be removed
+   * @param edges the list of edges to be removed
    * @return a new graph where the edges have been removed and in which the vertices remained intact
    */
   def removeEdges(edges: List[Edge[K, EV]]): Graph[K, VV, EV] = {
@@ -1016,7 +1016,7 @@ TypeInformation : ClassTag](jgraph: jg.Graph[K, VV, EV]) {
    * The {@link ReduceNeighborsFunction} combines a pair of neighbor vertex values
    * into one new value of the same type.
    * 
-   * @param reduceNeighborsFunction the reduce function to apply to the neighbors of each vertex.
+   * @param reduceEdgesFunction the reduce function to apply to the edges of each vertex.
    * @param direction the edge direction (in-, out-, all-)
    * @return a Dataset of Tuple2, with one tuple per vertex.
    * The first field of the Tuple2 is the vertex ID and the second field

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/ConnectedComponents.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/ConnectedComponents.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/ConnectedComponents.scala
index b3da520..75b793e 100644
--- a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/ConnectedComponents.scala
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/ConnectedComponents.scala
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.flink.graph.scala.example;
+package org.apache.flink.graph.scala.example
 
 import org.apache.flink.api.scala._
 import org.apache.flink.graph.scala._
@@ -32,7 +32,7 @@ import java.lang.Long
  * You can find all available library methods in [[org.apache.flink.graph.library]]. 
  * 
  * In particular, this example uses the
- * [[org.apache.flink.graph.library.ConnectedComponentsAlgorithm.GSAConnectedComponents]]
+ * [[org.apache.flink.graph.library.GSAConnectedComponents]]
  * library method to compute the connected components of the input graph.
  *
  * The input file is a plain text file and must be formatted as follows:
@@ -70,7 +70,7 @@ object ConnectedComponents {
   }
 
   private final class InitVertices extends MapFunction[Long, Long] {
-    override def map(id: Long) = {id}
+    override def map(id: Long) = id
   }
 
   // ***********************************************************************
@@ -87,19 +87,18 @@ object ConnectedComponents {
         if(args.length != 3) {
           System.err.println("Usage ConnectedComponents <edge path> <output path> " +
             "<num iterations>")
-          false
         }
         fileOutput = true
         edgesInputPath = args(0)
         outputPath = args(1)
-        maxIterations = (2).toInt
+        maxIterations = 2
       } else {
         System.out.println("Executing ConnectedComponents example with default parameters" +
           " and built-in default data.")
         System.out.println("  Provide parameters to read input data from files.")
         System.out.println("  See the documentation for the correct format of input files.")
         System.out.println("Usage ConnectedComponents <edge path> <output path> " +
-          "<num iterations>");
+          "<num iterations>")
       }
       true
     }

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GSASingleSourceShortestPaths.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GSASingleSourceShortestPaths.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GSASingleSourceShortestPaths.scala
index 2dc272c..68435ba 100644
--- a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GSASingleSourceShortestPaths.scala
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GSASingleSourceShortestPaths.scala
@@ -16,20 +16,15 @@
  * limitations under the License.
  */
 
-package org.apache.flink.graph.scala.example;
+package org.apache.flink.graph.scala.example
 
+import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.api.scala._
-import org.apache.flink.graph.scala._
-import org.apache.flink.types.NullValue
 import org.apache.flink.graph.Edge
-import org.apache.flink.api.common.functions.MapFunction
-import scala.collection.JavaConversions._
-import org.apache.flink.graph.scala.utils.Tuple3ToEdgeMap
 import org.apache.flink.graph.example.utils.SingleSourceShortestPathsData
-import org.apache.flink.graph.gsa.GatherFunction
-import org.apache.flink.graph.gsa.Neighbor
-import org.apache.flink.graph.gsa.SumFunction
-import org.apache.flink.graph.gsa.ApplyFunction
+import org.apache.flink.graph.gsa.{ApplyFunction, GatherFunction, Neighbor, SumFunction}
+import org.apache.flink.graph.scala._
+import org.apache.flink.graph.scala.utils.Tuple3ToEdgeMap
 
 /**
  * This example shows how to use Gelly's gather-sum-apply iterations.
@@ -121,20 +116,19 @@ object GSASingleSourceShortestPaths {
       if(args.length != 4) {
         System.err.println("Usage: SingleSourceShortestPaths <source vertex id>" +
           " <input edges path> <output path> <num iterations>")
-        false
       }
       fileOutput = true
       srcVertexId = args(0).toLong
       edgesInputPath = args(1)
       outputPath = args(2)
-      maxIterations = (3).toInt
+      maxIterations = 3
     } else {
       System.out.println("Executing Single Source Shortest Paths example "
         + "with default parameters and built-in default data.")
       System.out.println("  Provide parameters to read input data from files.")
       System.out.println("  See the documentation for the correct format of input files.")
       System.out.println("Usage: SingleSourceShortestPaths <source vertex id>" +
-        " <input edges path> <output path> <num iterations>");
+        " <input edges path> <output path> <num iterations>")
     }
     true
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GraphMetrics.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GraphMetrics.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GraphMetrics.scala
index 4eed824..1c3fcdd 100644
--- a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GraphMetrics.scala
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/GraphMetrics.scala
@@ -53,13 +53,13 @@ object GraphMetrics {
     val graph: Graph[Long, NullValue, NullValue] = Graph.fromDataSet(getEdgeDataSet(env), env)
 
     /** get the number of vertices **/
-    val numVertices = graph.numberOfVertices;
+    val numVertices = graph.numberOfVertices
 
     /** get the number of edges **/
-    val numEdges = graph.numberOfEdges;
+    val numEdges = graph.numberOfEdges
 
     /** compute the average node degree **/
-    val verticesWithDegrees = graph.getDegrees;
+    val verticesWithDegrees = graph.getDegrees
     val avgDegree = verticesWithDegrees.sum(1).map(in => (in._2 / numVertices).toDouble)
 
     /** find the vertex with the maximum in-degree **/
@@ -114,7 +114,7 @@ object GraphMetrics {
         (key: Long, out: Collector[Edge[Long, NullValue]]) => {
           val numOutEdges: Int = (Math.random() * (numVertices / 2)).toInt
           for ( i <- 0 to numOutEdges ) {
-            var target: Long = ((Math.random() * numVertices) + 1).toLong
+            val target: Long = ((Math.random() * numVertices) + 1).toLong
             new Edge[Long, NullValue](key, target, NullValue.getInstance())
           }
       })

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala
index 65a8e7f..7fc23c4 100644
--- a/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala
+++ b/flink-libraries/flink-gelly-scala/src/main/scala/org/apache/flink/graph/scala/example/SingleSourceShortestPaths.scala
@@ -16,11 +16,10 @@
  * limitations under the License.
  */
 
-package org.apache.flink.graph.scala.example;
+package org.apache.flink.graph.scala.example
 
 import org.apache.flink.api.scala._
 import org.apache.flink.graph.scala._
-import org.apache.flink.types.NullValue
 import org.apache.flink.graph.Edge
 import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.graph.spargel.VertexUpdateFunction
@@ -95,7 +94,7 @@ object SingleSourceShortestPaths {
     override def updateVertex(vertex: Vertex[Long, Double], inMessages: MessageIterator[Double]) {
       var minDistance = Double.MaxValue
       while (inMessages.hasNext) {
-        var msg = inMessages.next
+        val msg = inMessages.next
         if (msg < minDistance) {
           minDistance = msg
         }
@@ -115,7 +114,7 @@ object SingleSourceShortestPaths {
 
     override def sendMessages(vertex: Vertex[Long, Double]) {
       for (edge: Edge[Long, Double] <- getEdges) {
-        sendMessageTo(edge.getTarget(), vertex.getValue + edge.getValue)
+        sendMessageTo(edge.getTarget, vertex.getValue + edge.getValue)
       }
     }
   }
@@ -135,20 +134,19 @@ object SingleSourceShortestPaths {
       if(args.length != 4) {
         System.err.println("Usage: SingleSourceShortestPaths <source vertex id>" +
           " <input edges path> <output path> <num iterations>")
-        false
       }
       fileOutput = true
       srcVertexId = args(0).toLong
       edgesInputPath = args(1)
       outputPath = args(2)
-      maxIterations = (3).toInt
+      maxIterations = 3
     } else {
       System.out.println("Executing Single Source Shortest Paths example "
         + "with default parameters and built-in default data.")
       System.out.println("  Provide parameters to read input data from files.")
       System.out.println("  See the documentation for the correct format of input files.")
       System.out.println("Usage: SingleSourceShortestPaths <source vertex id>" +
-        " <input edges path> <output path> <num iterations>");
+        " <input edges path> <output path> <num iterations>")
     }
     true
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/GellyScalaAPICompletenessTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/GellyScalaAPICompletenessTest.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/GellyScalaAPICompletenessTest.scala
index 55faee3..d7ab1dd 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/GellyScalaAPICompletenessTest.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/GellyScalaAPICompletenessTest.scala
@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.flink.streaming.api.scala
+package org.apache.flink.graph.scala.test
 
 import java.lang.reflect.Method
 import org.apache.flink.graph.scala._

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/TestGraphUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/TestGraphUtils.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/TestGraphUtils.scala
index 1c2cf54..2fedfc7 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/TestGraphUtils.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/TestGraphUtils.scala
@@ -24,11 +24,11 @@ import org.apache.flink.graph.{Edge, Vertex}
 object TestGraphUtils {
 
     def getLongLongVertexData(env: ExecutionEnvironment): DataSet[Vertex[Long, Long]] = {
-        return env.fromCollection(getLongLongVertices)
+        env.fromCollection(getLongLongVertices)
     }
 
     def getLongLongEdgeData(env: ExecutionEnvironment): DataSet[Edge[Long, Long]] = {
-        return env.fromCollection(getLongLongEdges)
+        env.fromCollection(getLongLongEdges)
     }
 
     def getLongLongVertices: List[Vertex[Long, Long]] = {

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/DegreesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/DegreesITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/DegreesITCase.scala
index b347049..2a2b34e 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/DegreesITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/DegreesITCase.scala
@@ -22,10 +22,10 @@ import org.apache.flink.api.scala._
 import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])
@@ -36,33 +36,33 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testInDegrees {
+  def testInDegrees() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.inDegrees.collect().toList
+    val res = graph.inDegrees().collect().toList
     expectedResult = "(1,1)\n" + "(2,1)\n" + "(3,2)\n" + "(4,1)\n" + "(5,2)\n"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testOutDegrees {
+  def testOutDegrees() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.outDegrees.collect().toList
+    val res = graph.outDegrees().collect().toList
     expectedResult = "(1,2)\n" + "(2,1)\n" + "(3,2)\n" + "(4,1)\n" + "(5,1)\n"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testGetDegrees {
+  def testGetDegrees() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.getDegrees.collect().toList
+    val res = graph.getDegrees().collect().toList
     expectedResult = "(1,3)\n" + "(2,2)\n" + "(3,4)\n" + "(4,2)\n" + "(5,3)\n"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
index a963845..253040b 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
@@ -18,25 +18,20 @@
 
 package org.apache.flink.graph.scala.test.operations
 
+import java.io.{File, FileOutputStream, IOException, OutputStreamWriter}
+
+import com.google.common.base.Charsets
+import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.api.scala._
+import org.apache.flink.core.fs.{FileInputSplit, Path}
 import org.apache.flink.graph.scala._
-import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.apache.flink.types.NullValue
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
-import java.io.IOException
-import org.apache.flink.core.fs.FileInputSplit
-import java.io.File
-import java.io.OutputStreamWriter
-import java.io.FileOutputStream
-import java.io.FileOutputStream
-import com.google.common.base.Charsets
-import org.apache.flink.core.fs.Path
-import org.apache.flink.types.NullValue
-import org.apache.flink.api.common.functions.MapFunction
 
 @RunWith(classOf[Parameterized])
 class GraphCreationWithCsvITCase(mode: MultipleProgramsTestBase.TestExecutionMode) extends
@@ -46,7 +41,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testCsvWithValues {
+  def testCsvWithValues() {
     /*
      * Test with two Csv files, both vertices and edges have values
      */
@@ -61,14 +56,14 @@ MultipleProgramsTestBase(mode) {
         pathEdges = edgesSplit.getPath.toString,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,1,2,ot\n3,2,3,2,tt\n3,1,3,1,to\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvNoEdgeValues {
+  def testCsvNoEdgeValues() {
     /*
      * Test with two Csv files; edges have no values
      */
@@ -84,14 +79,14 @@ MultipleProgramsTestBase(mode) {
         hasEdgeValues = false,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,one,two,(null)\n3,2,three,two,(null)\n3,1,three,one,(null)\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvWithMapperValues {
+  def testCsvWithMapperValues() {
     /*
      * Test with edges Csv file and vertex mapper initializer
      */
@@ -104,14 +99,14 @@ MultipleProgramsTestBase(mode) {
         vertexValueInitializer = new VertexDoubleIdAssigner(),
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,1.0,2.0,12\n3,2,3.0,2.0,32\n3,1,3.0,1.0,31\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvNoVertexValues {
+  def testCsvNoVertexValues() {
     /*
      * Test with edges Csv file: no vertex values
      */
@@ -123,15 +118,15 @@ MultipleProgramsTestBase(mode) {
         pathEdges = edgesSplit.getPath.toString,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,(null),(null),12\n3,2,(null),(null),32\n" +
       "3,1,(null),(null),31\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvNoValues {
+  def testCsvNoValues() {
     /*
      * Test with edges Csv file: neither vertex nor edge values
      */
@@ -144,15 +139,15 @@ MultipleProgramsTestBase(mode) {
         hasEdgeValues = false,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,(null),(null),(null)\n" +
       "3,2,(null),(null),(null)\n3,1,(null),(null),(null)\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvOptionsVertices {
+  def testCsvOptionsVertices() {
     /*
      * Test the options for vertices: delimiters, comments, ignore first line.
      */
@@ -172,14 +167,14 @@ MultipleProgramsTestBase(mode) {
         pathEdges = edgesSplit.getPath.toString,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,1,2,ot\n3,2,3,2,tt\n3,1,3,1,to\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testCsvOptionsEdges {
+  def testCsvOptionsEdges() {
     /*
      * Test the options for edges: delimiters, comments, ignore first line.
      */
@@ -199,9 +194,9 @@ MultipleProgramsTestBase(mode) {
         pathEdges = edgesSplit.getPath.toString,
         env = env)
     
-    val result = graph.getTriplets.collect()
+    val result = graph.getTriplets().collect()
     expectedResult = "1,2,1,2,ot\n3,2,3,2,tt\n3,1,3,1,to\n"
-    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult);
+    TestBaseUtils.compareResultAsTuples(result.asJava, expectedResult)
   }
 
   @throws(classOf[IOException])
@@ -214,7 +209,7 @@ MultipleProgramsTestBase(mode) {
     wrt.close()
 
     new FileInputSplit(0, new Path(tempFile.toURI.toString), 0, tempFile.length,
-        Array("localhost"));
+        Array("localhost"))
     }
 
     final class VertexDoubleIdAssigner extends MapFunction[Long, Double] {

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphMutationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphMutationsITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphMutationsITCase.scala
index 4b776e2..f6acdc1 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphMutationsITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphMutationsITCase.scala
@@ -23,10 +23,10 @@ import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.graph.{Edge, Vertex}
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+import org.junit.Test
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])
@@ -37,7 +37,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddVertex {
+  def testAddVertex() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -50,7 +50,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddVertexExisting {
+  def testAddVertexExisting() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -62,7 +62,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddVertexNoEdges {
+  def testAddVertexNoEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -74,7 +74,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddVertices {
+  def testAddVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -88,7 +88,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddVerticesExisting {
+  def testAddVerticesExisting() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -102,7 +102,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveVertex {
+  def testRemoveVertex() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -114,7 +114,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveInvalidVertex {
+  def testRemoveInvalidVertex() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -127,7 +127,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveVertices {
+  def testRemoveVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -140,7 +140,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveValidAndInvalidVertex {
+  def testRemoveValidAndInvalidVertex() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -153,7 +153,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddEdge {
+  def testAddEdge() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -167,7 +167,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddEdges {
+  def testAddEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -181,7 +181,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddEdgesInvalidVertices {
+  def testAddEdgesInvalidVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -195,7 +195,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testAddExistingEdge {
+  def testAddExistingEdge() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -209,7 +209,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveEdge {
+  def testRemoveEdge() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -221,7 +221,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveInvalidEdge {
+  def testRemoveInvalidEdge() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -234,7 +234,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveEdges {
+  def testRemoveEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -247,7 +247,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testRemoveSameEdgeTwiceEdges {
+  def testRemoveSameEdgeTwiceEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphOperationsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphOperationsITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphOperationsITCase.scala
index 7f7ebc0..9d77e68 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphOperationsITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphOperationsITCase.scala
@@ -23,10 +23,9 @@ import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.graph.{Edge, Vertex}
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+import org.junit.Test
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])
@@ -37,11 +36,11 @@ MultipleProgramsTestBase(mode) {
     
   @Test
   @throws(classOf[Exception])
-  def testUndirected {
+  def testUndirected() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.getUndirected.getEdges.collect().toList;
+    val res = graph.getUndirected().getEdges.collect().toList
 
     expectedResult = "1,2,12\n" + "2,1,12\n" + "1,3,13\n" + "3,1,13\n" + "2,3,23\n" + "3,2," +
       "23\n" + "3,4,34\n" + "4,3,34\n" + "3,5,35\n" + "5,3,35\n" + "4,5,45\n" + "5,4,45\n" +
@@ -51,11 +50,11 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testReverse {
+  def testReverse() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.reverse().getEdges.collect().toList;
+    val res = graph.reverse().getEdges.collect().toList
 
     expectedResult = "2,1,12\n" + "3,1,13\n" + "3,2,23\n" + "4,3,34\n" + "5,3,35\n" + "5,4," +
       "45\n" + "1,5,51\n"
@@ -64,22 +63,22 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testSubGraph {
+  def testSubGraph() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
     val res = graph.subgraph(new FilterFunction[Vertex[Long, Long]] {
       @throws(classOf[Exception])
       def filter(vertex: Vertex[Long, Long]): Boolean = {
-        return (vertex.getValue > 2)
+        vertex.getValue > 2
       }
     }, new FilterFunction[Edge[Long, Long]] {
 
       @throws(classOf[Exception])
       override def filter(edge: Edge[Long, Long]): Boolean = {
-        return (edge.getValue > 34)
+        edge.getValue > 34
       }
-    }).getEdges.collect().toList;
+    }).getEdges.collect().toList
 
     expectedResult = "3,5,35\n" + "4,5,45\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -87,14 +86,14 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testSubGraphSugar {
+  def testSubGraphSugar() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
     val res = graph.subgraph(
       vertex => vertex.getValue > 2,
       edge => edge.getValue > 34
-    ).getEdges.collect().toList;
+    ).getEdges.collect().toList
 
     expectedResult = "3,5,35\n" + "4,5,45\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -102,7 +101,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testFilterOnVertices {
+  def testFilterOnVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -111,7 +110,7 @@ MultipleProgramsTestBase(mode) {
       def filter(vertex: Vertex[Long, Long]): Boolean = {
         vertex.getValue > 2
       }
-    }).getEdges.collect().toList;
+    }).getEdges.collect().toList
 
     expectedResult = "3,4,34\n" + "3,5,35\n" + "4,5,45\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -119,13 +118,13 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testFilterOnVerticesSugar {
+  def testFilterOnVerticesSugar() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
     val res = graph.filterOnVertices(
       vertex => vertex.getValue > 2
-    ).getEdges.collect().toList;
+    ).getEdges.collect().toList
 
     expectedResult = "3,4,34\n" + "3,5,35\n" + "4,5,45\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -133,7 +132,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testFilterOnEdges {
+  def testFilterOnEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -142,7 +141,7 @@ MultipleProgramsTestBase(mode) {
       def filter(edge: Edge[Long, Long]): Boolean = {
         edge.getValue > 34
       }
-    }).getEdges.collect().toList;
+    }).getEdges.collect().toList
 
     expectedResult = "3,5,35\n" + "4,5,45\n" + "5,1,51\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -150,13 +149,13 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testFilterOnEdgesSugar {
+  def testFilterOnEdgesSugar() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
     val res = graph.filterOnEdges(
       edge => edge.getValue > 34
-    ).getEdges.collect().toList;
+    ).getEdges.collect().toList
 
     expectedResult = "3,5,35\n" + "4,5,45\n" + "5,1,51\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
@@ -164,44 +163,44 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testNumberOfVertices {
+  def testNumberOfVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = env.fromElements(graph.numberOfVertices).collect().toList
+    val res = env.fromElements(graph.numberOfVertices()).collect().toList
     expectedResult = "5"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testNumberOfEdges {
+  def testNumberOfEdges() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = env.fromElements(graph.numberOfEdges).collect().toList
+    val res = env.fromElements(graph.numberOfEdges()).collect().toList
     expectedResult = "7"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testVertexIds {
+  def testVertexIds() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.getVertexIds.collect().toList
+    val res = graph.getVertexIds().collect().toList
     expectedResult = "1\n2\n3\n4\n5\n"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
   }
 
   @Test
   @throws(classOf[Exception])
-  def testEdgesIds {
+  def testEdgesIds() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.getEdgeIds.collect().toList
+    val res = graph.getEdgeIds().collect().toList
     expectedResult = "(1,2)\n" + "(1,3)\n" + "(2,3)\n" + "(3,4)\n" + "(3,5)\n" + "(4,5)\n" +
       "(5,1)\n"
     TestBaseUtils.compareResultAsText(res.asJava, expectedResult)
@@ -209,7 +208,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testUnion {
+  def testUnion() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -229,7 +228,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testDifference {
+  def testDifference() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -250,7 +249,7 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testDifferenceNoCommonVertices {
+  def testDifferenceNoCommonVertices() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
@@ -270,11 +269,11 @@ MultipleProgramsTestBase(mode) {
 
   @Test
   @throws(classOf[Exception])
-  def testTriplets {
+  def testTriplets() {
     val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
     val graph: Graph[Long, Long, Long] = Graph.fromDataSet(TestGraphUtils
       .getLongLongVertexData(env), TestGraphUtils.getLongLongEdgeData(env), env)
-    val res = graph.getTriplets.collect().toList
+    val res = graph.getTriplets().collect().toList
     expectedResult = "1,2,1,2,12\n" + "1,3,1,3,13\n" + "2,3,2,3,23\n" + "3,4,3,4,34\n" +
       "3,5,3,5,35\n" + "4,5,4,5,45\n" + "5,1,5,1,51\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithEdgesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithEdgesITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithEdgesITCase.scala
index 83fa61b..0a7f1b9 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithEdgesITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithEdgesITCase.scala
@@ -20,17 +20,16 @@ package org.apache.flink.graph.scala.test.operations
 
 import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.api.scala._
-import org.apache.flink.graph.Edge
+import org.apache.flink.graph.{Edge, EdgeJoinFunction}
 import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.graph.scala.utils.EdgeToTuple3Map
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
-import org.apache.flink.graph.EdgeJoinFunction
 
 @RunWith(classOf[Parameterized])
 class JoinWithEdgesITCase(mode: MultipleProgramsTestBase.TestExecutionMode) extends

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithVerticesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithVerticesITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithVerticesITCase.scala
index f2beb7b..5998270 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithVerticesITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/JoinWithVerticesITCase.scala
@@ -18,18 +18,17 @@
 
 package org.apache.flink.graph.scala.test.operations
 
-import org.apache.flink.api.common.functions.MapFunction
 import org.apache.flink.api.scala._
+import org.apache.flink.graph.VertexJoinFunction
 import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.graph.scala.utils.VertexToTuple2Map
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
-import org.apache.flink.graph.VertexJoinFunction
 
 @RunWith(classOf[Parameterized])
 class JoinWithVerticesITCase(mode: MultipleProgramsTestBase.TestExecutionMode) extends

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapEdgesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapEdgesITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapEdgesITCase.scala
index bdfd569..4c1d1f0 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapEdgesITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapEdgesITCase.scala
@@ -24,10 +24,10 @@ import org.apache.flink.graph.Edge
 import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapVerticesITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapVerticesITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapVerticesITCase.scala
index 2e51d90..a27b42c 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapVerticesITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/MapVerticesITCase.scala
@@ -24,10 +24,10 @@ import org.apache.flink.graph.Vertex
 import org.apache.flink.graph.scala._
 import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])
@@ -47,7 +47,7 @@ MultipleProgramsTestBase(mode) {
       "2,3\n" +
       "3,4\n" +
       "4,5\n" +
-      "5,6\n";
+      "5,6\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
   }
 
@@ -62,7 +62,7 @@ MultipleProgramsTestBase(mode) {
       "2,3\n" +
       "3,4\n" +
       "4,5\n" +
-      "5,6\n";
+      "5,6\n"
     TestBaseUtils.compareResultAsTuples(res.asJava, expectedResult)
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnEdgesMethodsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnEdgesMethodsITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnEdgesMethodsITCase.scala
index dcd1deb..6dda547 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnEdgesMethodsITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnEdgesMethodsITCase.scala
@@ -24,10 +24,10 @@ import org.apache.flink.graph.scala.test.TestGraphUtils
 import org.apache.flink.graph.scala.{EdgesFunction, EdgesFunctionWithVertexValue, Graph}
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
 import org.apache.flink.util.Collector
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnNeighborMethodsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnNeighborMethodsITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnNeighborMethodsITCase.scala
index aef5493..67e9b9a 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnNeighborMethodsITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/ReduceOnNeighborMethodsITCase.scala
@@ -24,10 +24,10 @@ import org.apache.flink.graph.scala.{NeighborsFunctionWithVertexValue, _}
 import org.apache.flink.graph.{Edge, EdgeDirection, ReduceNeighborsFunction, Vertex}
 import org.apache.flink.test.util.{MultipleProgramsTestBase, TestBaseUtils}
 import org.apache.flink.util.Collector
-import org.junit.rules.TemporaryFolder
+import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.{After, Before, Rule, Test}
+
 import _root_.scala.collection.JavaConverters._
 
 @RunWith(classOf[Parameterized])

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSP.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSP.java b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSP.java
index 947f343..f454376 100644
--- a/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSP.java
+++ b/flink-libraries/flink-gelly/src/main/java/org/apache/flink/graph/example/IncrementalSSSP.java
@@ -135,7 +135,7 @@ public class IncrementalSSSP implements ProgramDescription {
 	 *
 	 * @param edgeToBeRemoved
 	 * @param edgesInSSSP
-	 * @return
+	 * @return true or false
 	 */
 	public static boolean isInSSSP(final Edge<Long, Double> edgeToBeRemoved, DataSet<Edge<Long, Double>> edgesInSSSP) throws Exception {
 
@@ -154,9 +154,7 @@ public class IncrementalSSSP implements ProgramDescription {
 			if (inMessages.hasNext()) {
 				Long outDegree = getOutDegree() - 1;
 				// check if the vertex has another SP-Edge
-				if (outDegree > 0) {
-					// there is another shortest path from the source to this vertex
-				} else {
+				if (outDegree <= 0) {
 					// set own value to infinity
 					setNewVertexValue(Double.MAX_VALUE);
 				}

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSACompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSACompilerTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSACompilerTest.java
index 2ad203f..cd677b6 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSACompilerTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSACompilerTest.java
@@ -56,7 +56,7 @@ public class GSACompilerTest extends CompilerTestBase {
 			env.setParallelism(DEFAULT_PARALLELISM);
 			// compose test program
 			{
-				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple3<Long, Long, NullValue>(
+				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple3<>(
 						1L, 2L, NullValue.getInstance())).map(new Tuple3ToEdgeMap<Long, NullValue>());
 
 				Graph<Long, Long, NullValue> graph = Graph.fromDataSet(edges, new InitVertices(), env);
@@ -124,7 +124,7 @@ public class GSACompilerTest extends CompilerTestBase {
 		public Long gather(Neighbor<Long, NullValue> neighbor) {
 			return neighbor.getNeighborValue();
 		}
-	};
+	}
 
 	@SuppressWarnings("serial")
 	private static final class SelectMinId extends SumFunction<Long, NullValue, Long> {
@@ -132,7 +132,7 @@ public class GSACompilerTest extends CompilerTestBase {
 		public Long sum(Long newValue, Long currentValue) {
 			return Math.min(newValue, currentValue);
 		}
-	};
+	}
 
 	@SuppressWarnings("serial")
 	private static final class UpdateComponentId extends ApplyFunction<Long, Long, Long> {

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSATranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSATranslationTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSATranslationTest.java
index ced7508..2deebcb 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSATranslationTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/gsa/GSATranslationTest.java
@@ -71,7 +71,7 @@ public class GSATranslationTest {
 			// ------------ construct the test program ------------------
 			{
 
-				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple3<Long, Long, NullValue>(
+				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple3<>(
 						1L, 2L, NullValue.getInstance())).map(new Tuple3ToEdgeMap<Long, NullValue>());
 
 				Graph<Long, Long, NullValue> graph = Graph.fromDataSet(edges, new InitVertices(), env);
@@ -98,7 +98,7 @@ public class GSATranslationTest {
 			assertTrue(result instanceof DeltaIterationResultSet);
 			
 			DeltaIterationResultSet<?, ?> resultSet = (DeltaIterationResultSet<?, ?>) result;
-			DeltaIteration<?, ?> iteration = (DeltaIteration<?, ?>) resultSet.getIterationHead();
+			DeltaIteration<?, ?> iteration = resultSet.getIterationHead();
 			
 			// check the basic iteration properties
 			assertEquals(NUM_ITERATIONS, resultSet.getMaxIterations());
@@ -142,7 +142,7 @@ public class GSATranslationTest {
 		public Long gather(Neighbor<Long, NullValue> neighbor) {
 			return neighbor.getNeighborValue();
 		}
-	};
+	}
 
 	@SuppressWarnings("serial")
 	private static final class SelectMinId extends SumFunction<Long, NullValue, Long> {
@@ -150,7 +150,7 @@ public class GSATranslationTest {
 		public Long sum(Long newValue, Long currentValue) {
 			return Math.min(newValue, currentValue);
 		}
-	};
+	}
 
 	@SuppressWarnings("serial")
 	private static final class UpdateComponentId extends ApplyFunction<Long, Long, Long> {

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
index 7a8143a..335481f 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelCompilerTest.java
@@ -61,14 +61,14 @@ public class SpargelCompilerTest extends CompilerTestBase {
 			{
 
 				DataSet<Vertex<Long, Long>> initialVertices = env.fromElements(
-						new Tuple2<Long, Long>(1L, 1L), new Tuple2<Long, Long>(2L, 2L))
+						new Tuple2<>(1L, 1L), new Tuple2<>(2L, 2L))
 						.map(new Tuple2ToVertexMap<Long, Long>());
 
-				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple2<Long, Long>(1L, 2L))
+				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple2<>(1L, 2L))
 					.map(new MapFunction<Tuple2<Long,Long>, Edge<Long, NullValue>>() {
 
 						public Edge<Long, NullValue> map(Tuple2<Long, Long> edge) {
-							return new Edge<Long, NullValue>(edge.f0, edge.f1, NullValue.getInstance());
+							return new Edge<>(edge.f0, edge.f1, NullValue.getInstance());
 						}
 				});
 
@@ -143,14 +143,14 @@ public class SpargelCompilerTest extends CompilerTestBase {
 				DataSet<Long> bcVar = env.fromElements(1L);
 
 				DataSet<Vertex<Long, Long>> initialVertices = env.fromElements(
-						new Tuple2<Long, Long>(1L, 1L), new Tuple2<Long, Long>(2L, 2L))
+						new Tuple2<>(1L, 1L), new Tuple2<>(2L, 2L))
 						.map(new Tuple2ToVertexMap<Long, Long>());
 
-				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple2<Long, Long>(1L, 2L))
+				DataSet<Edge<Long, NullValue>> edges = env.fromElements(new Tuple2<>(1L, 2L))
 						.map(new MapFunction<Tuple2<Long,Long>, Edge<Long, NullValue>>() {
 
 							public Edge<Long, NullValue> map(Tuple2<Long, Long> edge) {
-								return new Edge<Long, NullValue>(edge.f0, edge.f1, NullValue.getInstance());
+								return new Edge<>(edge.f0, edge.f1, NullValue.getInstance());
 							}
 					});
 

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
index bb3a131..69aa99c 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/spargel/SpargelTranslationTest.java
@@ -52,7 +52,7 @@ public class SpargelTranslationTest {
 			final String BC_SET_MESSAGES_NAME = "borat messages";
 			
 			final String BC_SET_UPDATES_NAME = "borat updates";
-			;
+
 			final int NUM_ITERATIONS = 13;
 			
 			final int ITERATION_parallelism = 77;
@@ -68,16 +68,16 @@ public class SpargelTranslationTest {
 			// ------------ construct the test program ------------------
 			{
 				
-				DataSet<Tuple2<String, Double>> initialVertices = env.fromElements(new Tuple2<String, Double>("abc", 3.44));
+				DataSet<Tuple2<String, Double>> initialVertices = env.fromElements(new Tuple2<>("abc", 3.44));
 
-				DataSet<Tuple2<String, String>> edges = env.fromElements(new Tuple2<String, String>("a", "c"));
+				DataSet<Tuple2<String, String>> edges = env.fromElements(new Tuple2<>("a", "c"));
 
 				Graph<String, Double, NullValue> graph = Graph.fromTupleDataSet(initialVertices,
 						edges.map(new MapFunction<Tuple2<String,String>, Tuple3<String, String, NullValue>>() {
 
 							public Tuple3<String, String, NullValue> map(
 									Tuple2<String, String> edge) {
-								return new Tuple3<String, String, NullValue>(edge.f0, edge.f1, NullValue.getInstance());
+								return new Tuple3<>(edge.f0, edge.f1, NullValue.getInstance());
 							}
 						}), env);
 
@@ -101,7 +101,7 @@ public class SpargelTranslationTest {
 			assertTrue(result instanceof DeltaIterationResultSet);
 			
 			DeltaIterationResultSet<?, ?> resultSet = (DeltaIterationResultSet<?, ?>) result;
-			DeltaIteration<?, ?> iteration = (DeltaIteration<?, ?>) resultSet.getIterationHead();
+			DeltaIteration<?, ?> iteration = resultSet.getIterationHead();
 			
 			// check the basic iteration properties
 			assertEquals(NUM_ITERATIONS, resultSet.getMaxIterations());
@@ -139,7 +139,7 @@ public class SpargelTranslationTest {
 			final String BC_SET_MESSAGES_NAME = "borat messages";
 			
 			final String BC_SET_UPDATES_NAME = "borat updates";
-			;
+
 			final int NUM_ITERATIONS = 13;
 			
 			final int ITERATION_parallelism = 77;
@@ -154,16 +154,16 @@ public class SpargelTranslationTest {
 			// ------------ construct the test program ------------------
 			{
 
-				DataSet<Tuple2<String, Double>> initialVertices = env.fromElements(new Tuple2<String, Double>("abc", 3.44));
+				DataSet<Tuple2<String, Double>> initialVertices = env.fromElements(new Tuple2<>("abc", 3.44));
 
-				DataSet<Tuple2<String, String>> edges = env.fromElements(new Tuple2<String, String>("a", "c"));
+				DataSet<Tuple2<String, String>> edges = env.fromElements(new Tuple2<>("a", "c"));
 
 				Graph<String, Double, NullValue> graph = Graph.fromTupleDataSet(initialVertices,
 						edges.map(new MapFunction<Tuple2<String,String>, Tuple3<String, String, NullValue>>() {
 
 							public Tuple3<String, String, NullValue> map(
 									Tuple2<String, String> edge) {
-								return new Tuple3<String, String, NullValue>(edge.f0, edge.f1, NullValue.getInstance());
+								return new Tuple3<>(edge.f0, edge.f1, NullValue.getInstance());
 							}
 						}), env);
 
@@ -187,7 +187,7 @@ public class SpargelTranslationTest {
 			assertTrue(result instanceof DeltaIterationResultSet);
 			
 			DeltaIterationResultSet<?, ?> resultSet = (DeltaIterationResultSet<?, ?>) result;
-			DeltaIteration<?, ?> iteration = (DeltaIteration<?, ?>) resultSet.getIterationHead();
+			DeltaIteration<?, ?> iteration = resultSet.getIterationHead();
 			
 			// check the basic iteration properties
 			assertEquals(NUM_ITERATIONS, resultSet.getMaxIterations());

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
index 3fbd0bc..61fe0c2 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/CollectionModeSuperstepITCase.java
@@ -78,7 +78,7 @@ public class CollectionModeSuperstepITCase {
 	public static final class AssignOneMapper implements MapFunction<Vertex<Long, Long>, Long> {
 
 		public Long map(Vertex<Long, Long> value) {
-			return 1l;
+			return 1L;
 		}
 	}
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
index 67d32a8..1e44d5b 100644
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyConfigurationITCase.java
@@ -382,11 +382,10 @@ public class GatherSumApplyConfigurationITCase extends MultipleProgramsTestBase
 	private static final class FindAllReachableVertices extends SumFunction<HashSet<Long>, Long, HashSet<Long>> {
 		@Override
 		public HashSet<Long> sum(HashSet<Long> newSet, HashSet<Long> currentSet) {
-			HashSet<Long> set = currentSet;
 			for(Long l : newSet) {
-				set.add(l);
+				currentSet.add(l);
 			}
-			return set;
+			return currentSet;
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/ee648925/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
index 0213f02..039a05c 100755
--- a/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
+++ b/flink-libraries/flink-gelly/src/test/java/org/apache/flink/graph/test/GatherSumApplyITCase.java
@@ -79,7 +79,7 @@ public class GatherSumApplyITCase extends MultipleProgramsTestBase {
 				new InitMapperSSSP(), env);
 
         List<Vertex<Long, Double>> result = inputGraph.run(
-        		new GSASingleSourceShortestPaths<Long>(1l, 16)).collect();
+        		new GSASingleSourceShortestPaths<>(1L, 16)).collect();
 
 		expectedResult = "1,0.0\n" +
 				"2,12.0\n" +


Mime
View raw message