From 450b4241404055ed6638e354be421b83380827c5 Mon Sep 17 00:00:00 2001
From: zjureel <zjureel@gmail.com>
Date: Wed, 23 Aug 2017 10:30:31 +0800
Subject: [PATCH] [FLINK-6864] [core] Fix confusing "invalid POJO type"
 messages from TypeExtractor

This closes #4574
---
 docs/dev/types_serialization.md               |  3 +++
 .../api/java/typeutils/TypeExtractor.java     | 24 ++++++++++++++-----
 2 files changed, 21 insertions(+), 6 deletions(-)

diff --git a/docs/dev/types_serialization.md b/docs/dev/types_serialization.md
index 1f0c4661c3a..d865d8e3dd6 100644
--- 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
index 8ea2e1acfd9..4767838447f 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
@@ -1885,7 +1885,9 @@ public class TypeExtractor {
 			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,7 +1902,9 @@ public class TypeExtractor {
 
 		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);
 		}
 
@@ -1908,7 +1912,9 @@ public class TypeExtractor {
 		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 @@ public class TypeExtractor {
 		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 @@ public class TypeExtractor {
 				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;
 		}
 
-- 
GitLab