From c03237a860a4ee8f6461e1085b7e4c517b808ae5 Mon Sep 17 00:00:00 2001
From: Robert Metzger <rmetzger@apache.org>
Date: Fri, 21 Nov 2014 13:15:25 +0100
Subject: [PATCH] [FLINK-1221] Use only method name and location as default
 operator names

---
 .../org/apache/flink/api/java/ExecutionEnvironment.java     | 6 +++---
 .../src/main/java/org/apache/flink/api/java/Utils.java      | 6 +++++-
 .../java/org/apache/flink/api/java/operators/NamesTest.java | 4 ++--
 3 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index cae31573a30..38046793df7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/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());
 	}	
 	
 	// --------------------------------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index 462cf2c0935..1d510f9ec57 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/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());
 	}
 }
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operators/NamesTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operators/NamesTest.java
index 921d9cc048e..6e26e812ac5 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operators/NamesTest.java
+++ b/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;
 			}
-- 
GitLab