[FLINK-1333] Fixed getter/setter recognition for POJOs

This closes #271

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

@@ -989,12 +989,12 @@ public class TypeExtractor {
 			}
 			for(Method m : clazz.getMethods()) {
 				// check for getter
-				if(	// The name should be "get<FieldName>" or "<fieldName>" (for scala).
-					(m.getName().toLowerCase().equals("get"+fieldNameLow) || m.getName().toLowerCase().equals(fieldNameLow)) &&
+				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)) &&
 					// no arguments for the getter
 					m.getParameterTypes().length == 0 &&
 					// return type is same as field type (or the generic variant of it)
-					(m.getReturnType().equals( fieldType ) || (fieldTypeGeneric != null && m.getGenericReturnType().equals(fieldTypeGeneric)) )
+					(m.getGenericReturnType().equals( fieldType ) || (fieldTypeGeneric != null && m.getGenericReturnType().equals(fieldTypeGeneric)) )
 				) {
 					if(hasGetter) {
 						throw new IllegalStateException("Detected more than one getter");
@@ -1004,7 +1004,7 @@ public class TypeExtractor {
 				// check for setters (<FieldName>_$eq for scala)
 				if((m.getName().toLowerCase().equals("set"+fieldNameLow) || m.getName().toLowerCase().equals(fieldNameLow+"_$eq")) &&
 					m.getParameterTypes().length == 1 && // one parameter of the field's type
-					( m.getParameterTypes()[0].equals( fieldType ) || (fieldTypeGeneric != null && m.getGenericParameterTypes()[0].equals(fieldTypeGeneric) ) )&&
+					( m.getGenericParameterTypes()[0].equals( fieldType ) || (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/type/extractor/PojoTypeExtractionTest.java

@@ -19,6 +19,7 @@ package org.apache.flink.api.java.type.extractor;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 
@@ -138,6 +139,34 @@ public class PojoTypeExtractionTest {
 		}
 	}
 
+	public static class PojoWithGenericFields {
+		private Collection<String> users;
+		private boolean favorited;
+
+		public boolean isFavorited() {
+			return favorited;
+		}
+
+		public void setFavorited(boolean favorited) {
+			this.favorited = favorited;
+		}
+
+		public Collection<String> getUsers() {
+			return users;
+		}
+
+		public void setUsers(Collection<String> users) {
+			this.users = users;
+		}
+	}
+	@Test
+	public void testPojoWithGenericFields() {
+		TypeInformation<?> typeForClass = TypeExtractor.createTypeInfo(PojoWithGenericFields.class);
+
+		Assert.assertTrue(typeForClass instanceof PojoTypeInfo<?>);
+	}
+
+
 	// in this test, the location of the getters and setters is mixed across the type hierarchy.
 	public static class TypedPojoGetterSetterCheck extends GenericPojoGetterSetterCheck<String> {
 		public void setPackageProtected(String in) {