From b5d58934d7124e0076e588e74485a60e7c1f484b Mon Sep 17 00:00:00 2001
From: Ivan Mushketyk <ivan.mushketik@gmail.com>
Date: Mon, 1 Aug 2016 20:39:17 +0100
Subject: [PATCH] [FLINK-2090] [core] Truncate 'toString()' of
 CollectionInputFormat when the collection is huge

This closes #2323
---
 .../api/java/io/CollectionInputFormat.java    | 19 ++++++++-
 .../java/io/CollectionInputFormatTest.java    | 39 ++++++++++++++++++-
 2 files changed, 56 insertions(+), 2 deletions(-)

diff --git a/flink-java/src/main/java/org/apache/flink/api/java/io/CollectionInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/io/CollectionInputFormat.java
index 0dad9e355c8e0..90e6712fd3c8a 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/io/CollectionInputFormat.java
+++ b/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();
 	}
 	
 	// --------------------------------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
index 53523b01d7d2a..3bb40ebcb2f38 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
+++ b/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;
 	}