docs: public <T> DataSet<T> groupReduceOnNeighbors(NeighborsFunctionWithVertexValue<K, VV, EV, T> neighborsFunction,
  			EdgeDirection direction) throws IllegalArgumentException {
  		switch (direction) {
  		case IN:
  			// create <edge-sourceVertex> pairs
  			DataSet<Tuple2<Edge<K, EV>, Vertex<K, VV>>> edgesWithSources = edges
  					.join(this.vertices).where(0).equalTo(0).name("Edge with source vertex");
  			return vertices.coGroup(edgesWithSources)
  					.where(0).equalTo("f0.f1")
  					.with(new ApplyNeighborCoGroupFunction<>(neighborsFunction)).name("Neighbors function");
  		case OUT:
  			// create <edge-targetVertex> pairs
  			DataSet<Tuple2<Edge<K, EV>, Vertex<K, VV>>> edgesWithTargets = edges
  					.join(this.vertices).where(1).equalTo(0).name("Edge with target vertex");
  			return vertices.coGroup(edgesWithTargets)
  					.where(0).equalTo("f0.f0")
  					.with(new ApplyNeighborCoGroupFunction<>(neighborsFunction)).name("Neighbors function");
  		case ALL:
  			// create <edge-sourceOrTargetVertex> pairs
  			DataSet<Tuple3<K, Edge<K, EV>, Vertex<K, VV>>> edgesWithNeighbors = edges
  					.flatMap(new EmitOneEdgeWithNeighborPerNode<>()).name("Forward and reverse edges")
  					.join(this.vertices).where(1).equalTo(0)
  					.with(new ProjectEdgeWithNeighbor<>()).name("Edge with vertex");
  
  			return vertices.coGroup(edgesWithNeighbors)
  					.where(0).equalTo(0)
  					.with(new ApplyCoGroupFunctionOnAllNeighbors<>(neighborsFunction)).name("Neighbors function");
  		default:
  			throw new IllegalArgumentException("Illegal edge direction");
  		}
  	}