[FLINK-18484][core] Add length difference to arity exception in RowSe…

…rializer

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java

@@ -110,7 +110,8 @@ public Row copy(Row from) {
  int len = fieldSerializers.length;
 
  if (from.getArity() != len) {
- throw new RuntimeException("Row arity of from does not match serializers.");
+ throw new RuntimeException("Row arity of from (" + from.getArity() +
+ ") does not match this serializers field length (" + len + ").");
  }
 
  Row result = new Row(from.getKind(), len);
@@ -138,7 +139,7 @@ public Row copy(Row from, Row reuse) {
 
  if (from.getArity() != len || reuse.getArity() != len) {
  throw new RuntimeException(
- "Row arity of reuse or from is incompatible with this RowSerializer.");
+ "Row arity of reuse (" + reuse.getArity() + ") or from (" + from.getArity() + ") is incompatible with this serializers field length (" + len + ").");
  }
 
  reuse.setKind(from.getKind());
@@ -177,7 +178,7 @@ public void serialize(Row record, DataOutputView target) throws IOException {
  final int len = fieldSerializers.length;
 
  if (record.getArity() != len) {
- throw new RuntimeException("Row arity of from does not match serializers.");
+ throw new RuntimeException("Row arity of record (" + record.getArity() + ") does not match this serializers field length (" + len + ").");
  }
 
  // write bitmask
@@ -221,7 +222,7 @@ public Row deserialize(Row reuse, DataInputView source) throws IOException {
  final int len = fieldSerializers.length;
 
  if (reuse.getArity() != len) {
- throw new RuntimeException("Row arity of from does not match serializers.");
+ throw new RuntimeException("Row arity of reuse (" + reuse.getArity() + ") does not match this serializers field length (" + len + ").");
  }
 
  // read bitmask