[FLINK-2437] handle the case of a non-public default ctor in TypeExtractor.analyzePojo

Also changed some prints which printed the word "class" twice,
because class.toString also prepends it to the class name.

This closes #960.

docs/apis/programming_guide.md

@@ -1520,7 +1520,7 @@ Java and Scala classes are treated by Flink as a special POJO data type if they
 
 - The type of a field must be supported by Flink. At the moment, Flink uses [Avro](http://avro.apache.org) to serialize arbitrary objects (such as `Date`).
 
-Flink analyzes the structure of POJO types, i.e., it learns about the fields of a POJO. As a result POJO types are easier to use than general types. Moreover, they Flink can process POJOs more efficiently than general types.  
+Flink analyzes the structure of POJO types, i.e., it learns about the fields of a POJO. As a result POJO types are easier to use than general types. Moreover, Flink can process POJOs more efficiently than general types.
 
 The following example shows a simple POJO with two public fields.
 

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

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.java.typeutils;
 
+import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.GenericArrayType;
 import java.lang.reflect.Method;
@@ -1298,10 +1299,10 @@ public class TypeExtractor {
 				return true;
 			} else {
 				if(!hasGetter) {
-					LOG.debug("Class "+clazz+" does not contain a getter for field "+f.getName() );
+					LOG.debug(clazz+" does not contain a getter for field "+f.getName() );
 				}
 				if(!hasSetter) {
-					LOG.debug("Class "+clazz+" does not contain a setter for field "+f.getName() );
+					LOG.debug(clazz+" does not contain a setter for field "+f.getName() );
 				}
 				return false;
 			}
@@ -1323,7 +1324,7 @@ public class TypeExtractor {
 		
 		List<Field> fields = getAllDeclaredFields(clazz);
 		if(fields.size() == 0) {
-			LOG.info("No fields detected for class " + clazz + ". Cannot be used as a PojoType. Will be handled as GenericType");
+			LOG.info("No fields detected for " + clazz + ". Cannot be used as a PojoType. Will be handled as GenericType");
 			return new GenericTypeInfo<OUT>(clazz);
 		}
 
@@ -1331,7 +1332,7 @@ public class TypeExtractor {
 		for (Field field : fields) {
 			Type fieldType = field.getGenericType();
 			if(!isValidPojoField(field, clazz, typeHierarchy)) {
-				LOG.info("Class " + clazz + " is not a valid POJO type");
+				LOG.info(clazz + " is not a valid POJO type");
 				return null;
 			}
 			try {
@@ -1357,24 +1358,29 @@ public class TypeExtractor {
 		List<Method> methods = getAllDeclaredMethods(clazz);
 		for (Method method : methods) {
 			if (method.getName().equals("readObject") || method.getName().equals("writeObject")) {
-				LOG.info("Class "+clazz+" contains custom serialization methods we do not call.");
+				LOG.info(clazz+" contains custom serialization methods we do not call.");
 				return null;
 			}
 		}
 
 		// Try retrieving the default constructor, if it does not have one
 		// we cannot use this because the serializer uses it.
+		Constructor defaultConstructor = null;
 		try {
-			clazz.getDeclaredConstructor();
+			defaultConstructor = clazz.getDeclaredConstructor();
 		} catch (NoSuchMethodException e) {
 			if (clazz.isInterface() || Modifier.isAbstract(clazz.getModifiers())) {
-				LOG.info("Class " + clazz + " is abstract or an interface, having a concrete " +
+				LOG.info(clazz + " is abstract or an interface, having a concrete " +
 						"type can increase performance.");
 			} else {
-				LOG.info("Class " + clazz + " must have a default constructor to be used as a POJO.");
+				LOG.info(clazz + " must have a default constructor to be used as a POJO.");
 				return null;
 			}
 		}
+		if(defaultConstructor != null && !Modifier.isPublic(defaultConstructor.getModifiers())) {
+			LOG.info("The default constructor of " + clazz + " should be Public to be used as a POJO.");
+			return null;
+		}
 		
 		// everything is checked, we return the pojo
 		return pojoType;
@@ -1394,7 +1400,7 @@ public class TypeExtractor {
 					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 class "+clazz+"."
+					throw new RuntimeException("The field "+field+" is already contained in the hierarchy of the "+clazz+"."
 							+ "Please use unique field names through your classes hierarchy");
 				}
 				result.add(field);

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

@@ -313,6 +313,11 @@ public class TypeExtractorTest {
 		}
 	}
 
+	public static class PojoWithNonPublicDefaultCtor {
+		public int foo, bar;
+		PojoWithNonPublicDefaultCtor() {}
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	public void testPojo() {
@@ -345,6 +350,8 @@ public class TypeExtractorTest {
 		Assert.assertFalse(ti2.isTupleType());
 		Assert.assertTrue(ti2 instanceof PojoTypeInfo);
 		Assert.assertEquals(ti2.getTypeClass(), CustomType.class);
+
+		Assert.assertFalse(TypeExtractor.getForClass(PojoWithNonPublicDefaultCtor.class) instanceof PojoTypeInfo);
 	}