[FLINK-1133] TypeExtractor now also supports nested Tuple-input derivation

This closes #137

flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java

@@ -421,12 +421,9 @@ public class TypeExtractor {
 			Type[] tupleElements = ((ParameterizedType) inType).getActualTypeArguments();
 			// go thru all tuple elements and search for type variables
 			for(int i = 0; i < tupleElements.length; i++) {
-				if(tupleElements[i] instanceof TypeVariable) {
-					inType = materializeTypeVariable(returnTypeHierarchy, (TypeVariable<?>) tupleElements[i]);
-					info = findCorrespondingInfo(returnTypeVar, inType, ((TupleTypeInfo<?>) inTypeInfo).getTypeAt(i));
-					if(info != null) {
-						break;
-					}
+				info = createTypeInfoFromInput(returnTypeVar, returnTypeHierarchy, tupleElements[i], ((TupleTypeInfo<?>) inTypeInfo).getTypeAt(i));
+				if(info != null) {
+					break;
 				}
 			}
 		}

flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java

@@ -1470,4 +1470,24 @@ public class TypeExtractorTest {
 		Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tti.getTypeAt(0));
 		Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tti.getTypeAt(1));
 	}
+	
+	public static class DuplicateValueNested<T> implements MapFunction<Tuple1<Tuple1<T>>, Tuple2<T, T>> {
+		private static final long serialVersionUID = 1L;
+
+		@Override
+		public Tuple2<T, T> map(Tuple1<Tuple1<T>> vertex) {
+			return new Tuple2<T, T>(null, null);
+		}
+	}
+	
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testDuplicateValueNested() {
+		TypeInformation<?> ti = TypeExtractor.getMapReturnTypes((MapFunction) new DuplicateValueNested<String>(), TypeInfoParser.parse("Tuple1<Tuple1<String>>"));
+		Assert.assertTrue(ti.isTupleType());
+		Assert.assertEquals(2, ti.getArity());
+		TupleTypeInfo<?> tti = (TupleTypeInfo<?>) ti;
+		Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tti.getTypeAt(0));
+		Assert.assertEquals(BasicTypeInfo.STRING_TYPE_INFO, tti.getTypeAt(1));
+	}
 }