[FLINK-1182] Improve error messages for POJO type mismatches

flink-java/src/main/java/org/apache/flink/api/java/operators/Keys.java

@@ -246,8 +246,7 @@ public static <R> List<R> removeNullElementsFromList(List<R> in) {
  */
  public ExpressionKeys(String[] expressionsIn, TypeInformation<T> type) {
  if(!(type instanceof CompositeType<?>)) {
- throw new IllegalArgumentException("Type "+type+" is not a composite type. "
- + "Key expressions are only supported on POJO types and Tuples. "
+ throw new IllegalArgumentException("Key expressions are only supported on POJO types and Tuples. "
  + "A type is considered a POJO if all its fields are public, or have both getters and setters defined");
  }
  CompositeType<T> cType = (CompositeType<T>) type;

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

@@ -1031,7 +1031,7 @@ private <X> TypeInformation<X> analyzePojo(Class<X> clazz, ArrayList<Type> typeH
  } catch (InvalidTypesException e) {
  //pojoFields.add(new PojoField(field, new GenericTypeInfo( Object.class ))); // we need kryo to properly serialize this
  throw new InvalidTypesException("Flink is currently unable to serialize this type: "+fieldType+""
- + "\nThe system is using the Avro serializer which is not able to handle all types.", e);
+ + "\nThe system is internally using the Avro serializer which is not able to handle that type.", e);
  }
  }
 
@@ -1054,7 +1054,7 @@ private <X> TypeInformation<X> analyzePojo(Class<X> clazz, ArrayList<Type> typeH
  try {
  clazz.getDeclaredConstructor();
  } catch (NoSuchMethodException e) {
- LOG.warn("Class "+clazz+" does not have a default constructor. You can not use it as a POJO");
+ LOG.warn("Class " + clazz + " must have a default constructor to be used as a POJO.");
  return null;
  }