From 973170b1f04f151e80b88e3e6036fab80a3f707b Mon Sep 17 00:00:00 2001
From: twalthr <twalthr@apache.org>
Date: Wed, 18 Jan 2017 14:43:32 +0100
Subject: [PATCH] [FLINK-5549] [core] TypeExtractor fails with
 RuntimeException, but should use GenericTypeInfo

---
 .../api/java/typeutils/TypeExtractor.java     | 21 +++++++++++++------
 .../typeutils/PojoTypeExtractionTest.java     |  5 +++--
 2 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index 8bf286722f2eb..a2664f9e04eb0 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -893,7 +893,7 @@ else if (inTypeInfo instanceof PojoTypeInfo && isClassType(inType)) {
 			// build the entire type hierarchy for the pojo
 			getTypeHierarchy(inputTypeHierarchy, inType, Object.class);
 			// determine a field containing the type variable
-			List<Field> fields = getAllDeclaredFields(typeToClass(inType));
+			List<Field> fields = getAllDeclaredFields(typeToClass(inType), false);
 			for (Field field : fields) {
 				Type fieldType = field.getGenericType();
 				if (fieldType instanceof TypeVariable && sameTypeVars(returnTypeVar, materializeTypeVariable(inputTypeHierarchy, (TypeVariable<?>) fieldType))) {
@@ -1738,7 +1738,7 @@ else if (typeHierarchy.size() <= 1) {
 			getTypeHierarchy(typeHierarchy, clazz, Object.class);
 		}
 		
-		List<Field> fields = getAllDeclaredFields(clazz);
+		List<Field> fields = getAllDeclaredFields(clazz, false);
 		if (fields.size() == 0) {
 			LOG.info("No fields detected for " + clazz + ". Cannot be used as a PojoType. Will be handled as GenericType");
 			return new GenericTypeInfo<OUT>(clazz);
@@ -1803,12 +1803,17 @@ else if (typeHierarchy.size() <= 1) {
 	}
 
 	/**
-	 * recursively determine all declared fields
+	 * Recursively determine all declared fields
 	 * This is required because class.getFields() is not returning fields defined
 	 * in parent classes.
+	 *
+	 * @param clazz class to be analyzed
+	 * @param ignoreDuplicates if true, in case of duplicate field names only the lowest one
+	 *                            in a hierarchy will be returned; throws an exception otherwise
+	 * @return list of fields
 	 */
 	@PublicEvolving
-	public static List<Field> getAllDeclaredFields(Class<?> clazz) {
+	public static List<Field> getAllDeclaredFields(Class<?> clazz, boolean ignoreDuplicates) {
 		List<Field> result = new ArrayList<Field>();
 		while (clazz != null) {
 			Field[] fields = clazz.getDeclaredFields();
@@ -1817,8 +1822,12 @@ public static List<Field> getAllDeclaredFields(Class<?> clazz) {
 					continue; // we have no use for transient or static fields
 				}
 				if(hasFieldWithSameName(field.getName(), result)) {
-					throw new RuntimeException("The field "+field+" is already contained in the hierarchy of the "+clazz+"."
+					if (ignoreDuplicates) {
+						continue;
+					} else {
+						throw new InvalidTypesException("The field "+field+" is already contained in the hierarchy of the "+clazz+"."
 							+ "Please use unique field names through your classes hierarchy");
+					}
 				}
 				result.add(field);
 			}
@@ -1829,7 +1838,7 @@ public static List<Field> getAllDeclaredFields(Class<?> clazz) {
 
 	@PublicEvolving
 	public static Field getDeclaredField(Class<?> clazz, String name) {
-		for (Field field : getAllDeclaredFields(clazz)) {
+		for (Field field : getAllDeclaredFields(clazz, true)) {
 			if (field.getName().equals(name)) {
 				return field;
 			}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
index 2ca50814cf68a..ba09786ee36c2 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeExtractionTest.java
@@ -52,9 +52,10 @@ public static class HasDuplicateField extends WC {
 		private int count; // duplicate
 	}
 
-	@Test(expected=RuntimeException.class)
+	@Test
 	public void testDuplicateFieldException() {
-		TypeExtractor.createTypeInfo(HasDuplicateField.class);
+		TypeInformation<?> ti = TypeExtractor.createTypeInfo(HasDuplicateField.class);
+		Assert.assertTrue(ti instanceof GenericTypeInfo<?>);
 	}
 
 	// test with correct pojo types