[FLINK-6589] [core] Deserialize ArrayList with capacity of size+1 to …

…prevent growth.

This closes apache#3912.

flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/ListSerializer.java

@@ -82,7 +82,7 @@ public boolean isImmutableType() {
 	@Override
 	public TypeSerializer<List<T>> duplicate() {
 		TypeSerializer<T> duplicateElement = elementSerializer.duplicate();
-		return duplicateElement == elementSerializer ? this : new ListSerializer<T>(duplicateElement);
+		return duplicateElement == elementSerializer ? this : new ListSerializer<>(duplicateElement);
 	}
 
 	@Override
@@ -129,7 +129,8 @@ public void serialize(List<T> list, DataOutputView target) throws IOException {
 	@Override
 	public List<T> deserialize(DataInputView source) throws IOException {
 		final int size = source.readInt();
-		final List<T> list = new ArrayList<>(size);
+		// create new list with (size + 1) capacity to prevent expensive growth when a single element is added
+		final List<T> list = new ArrayList<>(size + 1);
 		for (int i = 0; i < size; i++) {
 			list.add(elementSerializer.deserialize(source));
 		}