[FLINK-1221] Use only method name and location as default operator names

flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java

@@ -426,7 +426,7 @@ public abstract class ExecutionEnvironment {
 		
 		TypeInformation<X> type = TypeExtractor.getForObject(firstValue);
 		CollectionInputFormat.checkCollection(data, type.getTypeClass());
-		return new DataSource<X>(this, new CollectionInputFormat<X>(data, type.createSerializer()), type, Utils.getCallLocationName(4));
+		return new DataSource<X>(this, new CollectionInputFormat<X>(data, type.createSerializer()), type, Utils.getCallLocationName());
 	}
 	
 	/**
@@ -572,7 +572,7 @@ public abstract class ExecutionEnvironment {
 	 * @see #fromParallelCollection(SplittableIterator, Class)
 	 */
 	public <X> DataSource<X> fromParallelCollection(SplittableIterator<X> iterator, TypeInformation<X> type) {
-		return fromParallelCollection(iterator, type, Utils.getCallLocationName(4));
+		return fromParallelCollection(iterator, type, Utils.getCallLocationName());
 	}
 	
 	// private helper for passing different call location names
@@ -589,7 +589,7 @@ public abstract class ExecutionEnvironment {
 	 * @return A DataSet, containing all number in the {@code [from, to]} interval.
 	 */
 	public DataSource<Long> generateSequence(long from, long to) {
-		return fromParallelCollection(new NumberSequenceIterator(from, to), BasicTypeInfo.LONG_TYPE_INFO, Utils.getCallLocationName(3));
+		return fromParallelCollection(new NumberSequenceIterator(from, to), BasicTypeInfo.LONG_TYPE_INFO, Utils.getCallLocationName());
 	}	
 	
 	// --------------------------------------------------------------------------------------------

flink-java/src/main/java/org/apache/flink/api/java/Utils.java

@@ -19,6 +19,8 @@
 package org.apache.flink.api.java;
 
 
+import org.apache.commons.lang3.StringUtils;
+
 public class Utils {
 
 	public static String getCallLocationName() {
@@ -30,6 +32,8 @@ public class Utils {
 		if(st.length < depth) { // we should not throw an out of bounds exception for this.
 			return "<unknown>";
 		}
-		return st[depth].toString();
+		String callLoc = st[depth].toString();
+		int idx = StringUtils.lastOrdinalIndexOf(callLoc, ".", 2); // second last occurrence of .
+		return callLoc.substring(idx+1, callLoc.length());
 	}
 }

flink-java/src/test/java/org/apache/flink/api/java/operators/NamesTest.java

@@ -60,7 +60,7 @@ public class NamesTest implements Serializable {
 			}
 		}).output(new DiscardingOuputFormat<String>());
 		JavaPlan plan = env.createProgramPlan();
-		testForName("Filter at org.apache.flink.api.java.operators.NamesTest.testDefaultName(NamesTest.java:54)", plan);
+		testForName("Filter at testDefaultName(NamesTest.java:54)", plan);
 	}
 	
 	@Test
@@ -102,7 +102,7 @@ public class NamesTest implements Serializable {
 			@Override
 			public boolean preVisit(Operator<?> visitable) {
 				if(visitable instanceof JoinOperatorBase) {
-					Assert.assertEquals("Join at org.apache.flink.api.java.operators.NamesTest.testJoinWith(NamesTest.java:92)", visitable.getName());
+					Assert.assertEquals("Join at testJoinWith(NamesTest.java:92)", visitable.getName());
 				}
 				return true;
 			}