[FLINK-1221] Use StackTraceElement methods instead of String parsing

This closes #231.

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

@@ -18,9 +18,6 @@
 
 package org.apache.flink.api.java;
 
-
-import org.apache.commons.lang3.StringUtils;
-
 public class Utils {
 
 	public static String getCallLocationName() {
@@ -28,12 +25,14 @@ public class Utils {
 	}
 
 	public static String getCallLocationName(int depth) {
-		StackTraceElement[] st = Thread.currentThread().getStackTrace();
-		if(st.length < depth) { // we should not throw an out of bounds exception for this.
+		StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
+
+		if (stackTrace.length < depth) {
 			return "<unknown>";
 		}
-		String callLoc = st[depth].toString();
-		int idx = StringUtils.lastOrdinalIndexOf(callLoc, ".", 2); // second last occurrence of .
-		return callLoc.substring(idx+1, callLoc.length());
+
+		StackTraceElement elem = stackTrace[depth];
+
+		return String.format("%s(%s:%d)", elem.getMethodName(), elem.getFileName(), elem.getLineNumber());
 	}
 }

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

@@ -15,6 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.flink.api.java.operators;
 
 import java.io.Serializable;
@@ -37,19 +38,18 @@ import org.apache.flink.util.Visitor;
 import org.junit.Assert;
 import org.junit.Test;
 
-
 /**
  * Test proper automated assignment of the transformation's name, if not set by the user.
  */
 public class NamesTest implements Serializable {
-	
+
 	@Test
 	public void testDefaultName() {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
+
 		DataSet<String> strs = env.fromCollection(Arrays.asList( new String[] {"a", "b"}));
-		
-		
+
+
 		// WARNING: The test will fail if this line is being moved down in the file (the line-number is hard-coded)
 		strs.filter(new FilterFunction<String>() {
 			private static final long serialVersionUID = 1L;
@@ -62,11 +62,11 @@ public class NamesTest implements Serializable {
 		JavaPlan plan = env.createProgramPlan();
 		testForName("Filter at testDefaultName(NamesTest.java:54)", plan);
 	}
-	
+
 	@Test
 	public void testGivenName() {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
+
 		DataSet<String> strs = env.fromCollection(Arrays.asList( new String[] {"a", "b"}));
 		strs.filter(new FilterFunction<String>() {
 			private static final long serialVersionUID = 1L;
@@ -78,25 +78,25 @@ public class NamesTest implements Serializable {
 		JavaPlan plan = env.createProgramPlan();
 		testForName("GivenName", plan);
 	}
-	
+
 	@Test
 	public void testJoinWith() {
 		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
+
 		List<Tuple1<String>> strLi = new ArrayList<Tuple1<String>>();
-		strLi.add(new Tuple1<String>("a")); 
+		strLi.add(new Tuple1<String>("a"));
 		strLi.add(new Tuple1<String>("b"));
-		
+
 		DataSet<Tuple1<String>> strs = env.fromCollection(strLi);
 		DataSet<Tuple1<String>> strs1 = env.fromCollection(strLi);
 		strs.join(strs1).where(0).equalTo(0).with(new FlatJoinFunction<Tuple1<String>, Tuple1<String>, String>() {
 			@Override
 			public void join(Tuple1<String> first, Tuple1<String> second,
-					Collector<String> out) throws Exception {
+							 Collector<String> out) throws Exception {
 				//
 			}
 		})
-		.output(new DiscardingOuputFormat<String>());
+				.output(new DiscardingOuputFormat<String>());
 		JavaPlan plan = env.createProgramPlan();
 		plan.accept(new Visitor<Operator<?>>() {
 			@Override
@@ -110,40 +110,23 @@ public class NamesTest implements Serializable {
 			public void postVisit(Operator<?> visitable) {}
 		});
 	}
-	
+
 	private static void testForName(final String expected, JavaPlan plan) {
 		plan.accept(new Visitor<Operator<?>>() {
 			@Override
 			public boolean preVisit(Operator<?> visitable) {
-				if(visitable instanceof PlanFilterOperator<?>) { 
+				if(visitable instanceof PlanFilterOperator<?>) {
 					// cast is actually not required. Its just a check for the right element
 					PlanFilterOperator<?> filterOp = (PlanFilterOperator<?>) visitable;
 					Assert.assertEquals(expected, filterOp.getName());
 				}
 				return true;
 			}
-			
+
 			@Override
 			public void postVisit(Operator<?> visitable) {
 				//
 			}
 		});
 	}
-	
-	/*public static void main(String[] args) throws Exception {
-		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		
-		DataSource<String> strs = env.fromParallelCollection(null, String.class);
-		strs.output(new DiscardingOuputFormat<String>());
-		JavaPlan plan = env.createProgramPlan();
-		plan.accept(new Visitor<Operator<?>>() {
-			@Override
-			public boolean preVisit(Operator<?> visitable) {
-				System.err.println("vis = "+visitable);
-				return true;
-			}
-			@Override
-			public void postVisit(Operator<?> visitable) {}
-		});
-	} */
 }