docs: @SuppressWarnings("serial")
  	private DataSet<Vertex<K, VV>> createResultVerticesWithDegrees(Graph<K, VV, EV> graph, EdgeDirection messagingDirection,
  			TypeInformation<Tuple2<K, Message>> messageTypeInfo, DataSet<LongValue> numberOfVertices) {
  
  		DataSet<Tuple2<K, Message>> messages;
  
  		this.gatherFunction.setOptDegrees(this.configuration.isOptDegrees());
  
  		DataSet<Tuple2<K, LongValue>> inDegrees = graph.inDegrees();
  		DataSet<Tuple2<K, LongValue>> outDegrees = graph.outDegrees();
  
  		DataSet<Tuple3<K, LongValue, LongValue>> degrees = inDegrees.join(outDegrees).where(0).equalTo(0)
  				.with(new FlatJoinFunction<Tuple2<K, LongValue>, Tuple2<K, LongValue>, Tuple3<K, LongValue, LongValue>>() {
  
  					@Override
  					public void join(Tuple2<K, LongValue> first, Tuple2<K, LongValue> second, Collector<Tuple3<K, LongValue, LongValue>> out) {
  						out.collect(new Tuple3<>(first.f0, first.f1, second.f1));
  					}
  				}).withForwardedFieldsFirst("f0;f1").withForwardedFieldsSecond("f1");
  
  		DataSet<Vertex<K, Tuple3<VV, LongValue, LongValue>>> verticesWithDegrees = initialVertices
  				.join(degrees).where(0).equalTo(0)
  				.with(new FlatJoinFunction<Vertex<K, VV>, Tuple3<K, LongValue, LongValue>, Vertex<K, Tuple3<VV, LongValue, LongValue>>>() {
  					@Override
  					public void join(Vertex<K, VV> vertex, Tuple3<K, LongValue, LongValue> degrees,
  									Collector<Vertex<K, Tuple3<VV, LongValue, LongValue>>> out) throws Exception {
  						out.collect(new Vertex<>(vertex.getId(),
  									new Tuple3<>(vertex.getValue(), degrees.f1, degrees.f2)));
  					}
  				}).withForwardedFieldsFirst("f0");
  
  		// add type info
  		TypeInformation<Vertex<K, Tuple3<VV, LongValue, LongValue>>> vertexTypes = verticesWithDegrees.getType();
  
  		final DeltaIteration<Vertex<K, Tuple3<VV, LongValue, LongValue>>, Vertex<K, Tuple3<VV, LongValue, LongValue>>> iteration =
  				verticesWithDegrees.iterateDelta(verticesWithDegrees, this.maximumNumberOfIterations, 0);
  				setUpIteration(iteration);
  
  		switch (messagingDirection) {
  			case IN:
  				messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices);
  				break;
  			case OUT:
  				messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices);
  				break;
  			case ALL:
  				messages = buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 1, 0, numberOfVertices)
  						.union(buildScatterFunctionVerticesWithDegrees(iteration, messageTypeInfo, 0, 0, numberOfVertices));
  				break;
  			default:
  				throw new IllegalArgumentException("Illegal edge direction");
  		}
  
  		@SuppressWarnings({ "unchecked", "rawtypes" })
  		GatherUdf<K, Tuple3<VV, LongValue, LongValue>, Message> updateUdf =
  				new GatherUdfVVWithDegrees(gatherFunction, vertexTypes);
  
  		// build the update function (co group)
  		CoGroupOperator<?, ?, Vertex<K, Tuple3<VV, LongValue, LongValue>>> updates =
  				messages.coGroup(iteration.getSolutionSet()).where(0).equalTo(0).with(updateUdf);
  
  		if (this.configuration != null && this.configuration.isOptNumVertices()) {
  			updates = updates.withBroadcastSet(numberOfVertices, "number of vertices");
  		}
  
  		configureUpdateFunction(updates);
  
  		return iteration.closeWith(updates, updates).map(
  				new MapFunction<Vertex<K, Tuple3<VV, LongValue, LongValue>>, Vertex<K, VV>>() {
  
  					public Vertex<K, VV> map(Vertex<K, Tuple3<VV, LongValue, LongValue>> vertex) {
  						return new Vertex<>(vertex.getId(), vertex.getValue().f0);
  					}
  				});
  	}