[FLINK-2874] Fix Avro getter/setter recognition

This closes #1252

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

@@ -30,6 +30,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.commons.lang3.ClassUtils;
 import org.apache.flink.api.common.functions.CoGroupFunction;
 import org.apache.flink.api.common.functions.CrossFunction;
 import org.apache.flink.api.common.functions.FlatJoinFunction;
@@ -1299,22 +1300,26 @@ public class TypeExtractor {
 			return true;
 		} else {
 			boolean hasGetter = false, hasSetter = false;
-			final String fieldNameLow = f.getName().toLowerCase();
-			
+			final String fieldNameLow = f.getName().toLowerCase().replaceAll("_", "");
+
 			Type fieldType = f.getGenericType();
+			Class<?> fieldTypeWrapper = ClassUtils.primitiveToWrapper(f.getType());
+
 			TypeVariable<?> fieldTypeGeneric = null;
 			if(fieldType instanceof TypeVariable) {
 				fieldTypeGeneric = (TypeVariable<?>) fieldType;
 				fieldType = materializeTypeVariable(typeHierarchy, (TypeVariable<?>)fieldType);
 			}
 			for(Method m : clazz.getMethods()) {
+				final String methodNameLow = m.getName().toLowerCase().replaceAll("_", "");
+
 				// check for getter
 				if(	// The name should be "get<FieldName>" or "<fieldName>" (for scala) or "is<fieldName>" for boolean fields.
-					(m.getName().toLowerCase().equals("get"+fieldNameLow) || m.getName().toLowerCase().equals("is"+fieldNameLow) || m.getName().toLowerCase().equals(fieldNameLow)) &&
+					(methodNameLow.equals("get"+fieldNameLow) || methodNameLow.equals("is"+fieldNameLow) || methodNameLow.equals(fieldNameLow)) &&
 					// no arguments for the getter
 					m.getParameterTypes().length == 0 &&
 					// return type is same as field type (or the generic variant of it)
-					(m.getGenericReturnType().equals( fieldType ) || (fieldTypeGeneric != null && m.getGenericReturnType().equals(fieldTypeGeneric)) )
+					(m.getGenericReturnType().equals( fieldType ) || (fieldTypeWrapper != null && m.getReturnType().equals( fieldTypeWrapper )) || (fieldTypeGeneric != null && m.getGenericReturnType().equals(fieldTypeGeneric)) )
 				) {
 					if(hasGetter) {
 						throw new IllegalStateException("Detected more than one getter");
@@ -1322,9 +1327,9 @@ public class TypeExtractor {
 					hasGetter = true;
 				}
 				// check for setters (<FieldName>_$eq for scala)
-				if((m.getName().toLowerCase().equals("set"+fieldNameLow) || m.getName().toLowerCase().equals(fieldNameLow+"_$eq")) &&
+				if((methodNameLow.equals("set"+fieldNameLow) || methodNameLow.equals(fieldNameLow+"_$eq")) &&
 					m.getParameterTypes().length == 1 && // one parameter of the field's type
-					( m.getGenericParameterTypes()[0].equals( fieldType ) || (fieldTypeGeneric != null && m.getGenericParameterTypes()[0].equals(fieldTypeGeneric) ) )&&
+					(m.getGenericParameterTypes()[0].equals( fieldType ) || (fieldTypeWrapper != null && m.getParameterTypes()[0].equals( fieldTypeWrapper )) || (fieldTypeGeneric != null && m.getGenericParameterTypes()[0].equals(fieldTypeGeneric) ) )&&
 					// return type is void.
 					m.getReturnType().equals(Void.TYPE)
 				) {

flink-java/src/test/java/org/apache/flink/api/java/typeutils/PojoTypeInfoTest.java

@@ -74,11 +74,25 @@ public class PojoTypeInfoTest {
 
 		assertEquals(pojoTypeInfo, deserializedPojoTypeInfo);
 	}
-	
+
+	@Test
+	public void testPrimitivePojo() {
+		TypeInformation<PrimitivePojo> info1 = TypeExtractor.getForClass(PrimitivePojo.class);
+
+		assertTrue(info1 instanceof PojoTypeInfo);
+	}
+
+	@Test
+	public void testUnderscorePojo() {
+		TypeInformation<UnderscorePojo> info1 = TypeExtractor.getForClass(UnderscorePojo.class);
+
+		assertTrue(info1 instanceof PojoTypeInfo);
+	}
+
 	public static final class TestPojo {
 		
 		public int someInt;
-		
+
 		private String aString;
 		
 		public Double[] doubleArray;
@@ -110,4 +124,30 @@ public class PojoTypeInfoTest {
 			return aString;
 		}
 	}
+
+	public static final class PrimitivePojo {
+
+		private int someInt;
+
+		public void setSomeInt(Integer someInt) {
+			this.someInt = someInt;
+		}
+
+		public Integer getSomeInt() {
+			return this.someInt;
+		}
+	}
+
+	public static final class UnderscorePojo {
+
+		private int some_int;
+
+		public void setSomeInt(int some_int) {
+			this.some_int = some_int;
+		}
+
+		public Integer getSomeInt() {
+			return this.some_int;
+		}
+	}
 }