[FLINK-2090] [core] Truncate 'toString()' of CollectionInputFormat wh…

…en the collection is huge

This closes apache#2323

flink-java/src/main/java/org/apache/flink/api/java/io/CollectionInputFormat.java

@@ -41,6 +41,7 @@
 public class CollectionInputFormat<T> extends GenericInputFormat<T> implements NonParallelInput {
 
 	private static final long serialVersionUID = 1L;
+	private static final int MAX_TO_STRING_LEN = 100;
 
 	private TypeSerializer<T> serializer;
 
@@ -117,7 +118,23 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
 
 	@Override
 	public String toString() {
-		return this.dataSet.toString();
+		StringBuilder sb = new StringBuilder();
+		sb.append('[');
+
+		int num = 0;
+		for (T e : dataSet) {
+			sb.append(e);
+			if (num != dataSet.size() - 1) {
+				sb.append(", ");
+				if (sb.length() > MAX_TO_STRING_LEN) {
+					sb.append("...");
+					break;
+				}
+			}
+			num++;
+		}
+		sb.append(']');
+		return sb.toString();
 	}
 
 	// --------------------------------------------------------------------------------------------

flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java

@@ -76,6 +76,13 @@ public boolean equals(Object obj) {
 		public int hashCode() {
 			return id;
 		}
+
+		@Override
+		public String toString() {
+			return "ElementType{" +
+				"id=" + id +
+				'}';
+		}
 	}
 
 	@Test
@@ -253,7 +260,37 @@ public void testDeserializationFailure() {
 			fail(e.getMessage());
 		}
 	}
-
+
+	@Test
+	public void testToStringOnSmallCollection() {
+		ArrayList<ElementType> smallList = new ArrayList<>();
+		smallList.add(new ElementType(1));
+		smallList.add(new ElementType(2));
+		CollectionInputFormat<ElementType> inputFormat = new CollectionInputFormat<>(
+			smallList,
+			new TestSerializer(true, false)
+		);
+
+		assertEquals("[ElementType{id=1}, ElementType{id=2}]", inputFormat.toString());
+	}
+
+	@Test
+	public void testToStringOnBigCollection() {
+		ArrayList<ElementType> list = new ArrayList<>();
+		for (int i = 0; i < 10; i++) {
+			list.add(new ElementType(i));
+		}
+		CollectionInputFormat<ElementType> inputFormat = new CollectionInputFormat<>(
+			list,
+			new TestSerializer(true, false)
+		);
+
+		assertEquals(
+			"[ElementType{id=0}, ElementType{id=1}, ElementType{id=2}, " +
+			"ElementType{id=3}, ElementType{id=4}, ElementType{id=5}, ...]",
+			inputFormat.toString());
+	}
+
 	private static class TestException extends IOException{
 		private static final long serialVersionUID = 1L;
 	}