[FLINK-6864] [core] Fix confusing "invalid POJO type" messages from T…

…ypeExtractor This closes apache#4574
JoeMalt · Nov 28, 2017 · 450b424 · 450b424
1 parent 53f2c1c
commit 450b424
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/docs/dev/types_serialization.md b/docs/dev/types_serialization.md
@@ -115,6 +115,9 @@ conditions are fulfilled:
   or have a public getter- and a setter- method that follows the Java beans
   naming conventions for getters and setters.
 
+Note that when a user-defined data type can't be recognized as a POJO type, it must be processed as GenericType and
+serialized with Kryo.
+
 
 #### Creating a TypeInformation or TypeSerializer
 

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
@@ -1885,7 +1885,9 @@ protected <OUT, IN1, IN2> TypeInformation<OUT> analyzePojo(Class<OUT> clazz, Arr
 			ParameterizedType parameterizedType, TypeInformation<IN1> in1Type, TypeInformation<IN2> in2Type) {
 
 		if (!Modifier.isPublic(clazz.getModifiers())) {
-			LOG.info("Class " + clazz.getName() + " is not public, cannot treat it as a POJO type. Will be handled as GenericType");
+			LOG.info("Class " + clazz.getName() + " is not public so it cannot be used as a POJO type " +
+				"and must be processed as GenericType. Please read the Flink documentation " +
+				"on \"Data Types & Serialization\" for details of the effect on performance.");
 			return new GenericTypeInfo<OUT>(clazz);
 		}
 
@@ -1900,15 +1902,19 @@ else if (typeHierarchy.size() <= 1) {
 
 		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");
+			LOG.info("No fields were detected for " + clazz + " so it cannot be used as a POJO type " +
+				"and must be processed as GenericType. Please read the Flink documentation " +
+				"on \"Data Types & Serialization\" for details of the effect on performance.");
 			return new GenericTypeInfo<OUT>(clazz);
 		}
 
 		List<PojoField> pojoFields = new ArrayList<PojoField>();
 		for (Field field : fields) {
 			Type fieldType = field.getGenericType();
 			if(!isValidPojoField(field, clazz, typeHierarchy)) {
-				LOG.info(clazz + " is not a valid POJO type because not all fields are valid POJO fields.");
+				LOG.info("Class " + clazz + " cannot be used as a POJO type because not all fields are valid POJO fields, " +
+					"and must be processed as GenericType. Please read the Flink documentation " +
+					"on \"Data Types & Serialization\" for details of the effect on performance.");
 				return null;
 			}
 			try {
@@ -1934,7 +1940,9 @@ else if (typeHierarchy.size() <= 1) {
 		List<Method> methods = getAllDeclaredMethods(clazz);
 		for (Method method : methods) {
 			if (method.getName().equals("readObject") || method.getName().equals("writeObject")) {
-				LOG.info(clazz+" contains custom serialization methods we do not call.");
+				LOG.info("Class " + clazz + " contains custom serialization methods we do not call, so it cannot be used as a POJO type " +
+					"and must be processed as GenericType. Please read the Flink documentation " +
+					"on \"Data Types & Serialization\" for details of the effect on performance.");
 				return null;
 			}
 		}
@@ -1949,12 +1957,16 @@ else if (typeHierarchy.size() <= 1) {
 				LOG.info(clazz + " is abstract or an interface, having a concrete " +
 						"type can increase performance.");
 			} else {
-				LOG.info(clazz + " must have a default constructor to be used as a POJO.");
+				LOG.info(clazz + " is missing a default constructor so it cannot be used as a POJO type " +
+					"and must be processed as GenericType. Please read the Flink documentation " +
+					"on \"Data Types & Serialization\" for details of the effect on performance.");
 				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.");
+			LOG.info("The default constructor of " + clazz + " is not Public so it cannot be used as a POJO type " +
+				"and must be processed as GenericType. Please read the Flink documentation " +
+				"on \"Data Types & Serialization\" for details of the effect on performance.");
 			return null;
 		}