flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [2/4] flink git commit: [FLINK-2457] [core] Integrate Tuple0 - extended TupleType helper classes to handle Tuple0 - extended TupleSerializer to handle Tuple0 - included Tuple0 into JUnit tests - simplified Receiver.createTuple(int)
Date Fri, 14 Aug 2015 18:38:00 GMT
[FLINK-2457] [core] Integrate Tuple0
 - extended TupleType helper classes to handle Tuple0
 - extended TupleSerializer to handle Tuple0
 - included Tuple0 into JUnit tests
 - simplified Receiver.createTuple(int)

This closes #983


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/fab9ce5d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/fab9ce5d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/fab9ce5d

Branch: refs/heads/master
Commit: fab9ce5d87976d22c2fec0cfb732fb6526d6ee15
Parents: 852d19c
Author: mjsax <mjsax@informatik.hu-berlin.de>
Authored: Sun Aug 2 21:38:31 2015 +0200
Committer: Stephan Ewen <sewen@apache.org>
Committed: Fri Aug 14 18:07:11 2015 +0200

----------------------------------------------------------------------
 .../org/apache/flink/api/java/tuple/Tuple.java  |  6 +-
 .../org/apache/flink/api/java/tuple/Tuple0.java | 60 ++++++++++++-
 .../flink/api/java/tuple/TupleGenerator.java    | 11 +--
 .../api/java/tuple/builder/Tuple0Builder.java   | 46 ++++++++++
 .../api/java/tuple/builder/Tuple10Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple11Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple12Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple13Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple14Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple15Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple16Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple17Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple18Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple19Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple1Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple20Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple21Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple22Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple23Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple24Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple25Builder.java  |  4 +-
 .../api/java/tuple/builder/Tuple2Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple3Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple4Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple5Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple6Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple7Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple8Builder.java   |  4 +-
 .../api/java/tuple/builder/Tuple9Builder.java   |  4 +-
 .../flink/api/java/typeutils/TupleTypeInfo.java |  8 +-
 .../api/java/typeutils/TupleTypeInfoBase.java   | 18 ++--
 .../flink/api/java/typeutils/TypeExtractor.java | 22 +++--
 .../api/java/typeutils/TypeInfoParser.java      | 35 +++++---
 .../typeutils/runtime/Tuple0Serializer.java     | 91 ++++++++++++++++++++
 .../java/typeutils/runtime/TupleSerializer.java |  2 +-
 .../java/type/extractor/TypeExtractorTest.java  | 22 +++++
 .../api/java/typeutils/TypeInfoParserTest.java  |  8 ++
 .../typeutils/runtime/TupleSerializerTest.java  | 16 +++-
 .../api/java/common/streaming/Receiver.java     | 70 +++------------
 39 files changed, 363 insertions(+), 152 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
index 145d215..3b07aed 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
@@ -93,10 +93,10 @@ public abstract class Tuple implements java.io.Serializable {
 	 */
 	@SuppressWarnings("unchecked")
 	public static Class<? extends Tuple> getTupleClass(int arity) {
-		if (arity < 1 || arity > MAX_ARITY) {
+		if (arity < 0 || arity > MAX_ARITY) {
 			throw new IllegalArgumentException("The tuple arity must be in [0, " + MAX_ARITY + "].");
 		}
-		return (Class<? extends Tuple>) CLASSES[arity - 1];
+		return (Class<? extends Tuple>) CLASSES[arity];
 	}
 	
 	// --------------------------------------------------------------------------------------------	
@@ -106,7 +106,7 @@ public abstract class Tuple implements java.io.Serializable {
 	// BEGIN_OF_TUPLE_DEPENDENT_CODE	
 	// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
 	private static final Class<?>[] CLASSES = new Class<?>[] {
-		Tuple1.class, Tuple2.class, Tuple3.class, Tuple4.class, Tuple5.class, Tuple6.class, Tuple7.class, Tuple8.class, Tuple9.class, Tuple10.class, Tuple11.class, Tuple12.class, Tuple13.class, Tuple14.class, Tuple15.class, Tuple16.class, Tuple17.class, Tuple18.class, Tuple19.class, Tuple20.class, Tuple21.class, Tuple22.class, Tuple23.class, Tuple24.class, Tuple25.class
+		Tuple0.class, Tuple1.class, Tuple2.class, Tuple3.class, Tuple4.class, Tuple5.class, Tuple6.class, Tuple7.class, Tuple8.class, Tuple9.class, Tuple10.class, Tuple11.class, Tuple12.class, Tuple13.class, Tuple14.class, Tuple15.class, Tuple16.class, Tuple17.class, Tuple18.class, Tuple19.class, Tuple20.class, Tuple21.class, Tuple22.class, Tuple23.class, Tuple24.class, Tuple25.class
 	};
 	// END_OF_TUPLE_DEPENDENT_CODE
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple0.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple0.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple0.java
index 82209bf..c2db0bc 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple0.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/Tuple0.java
@@ -17,23 +17,75 @@
  */
 package org.apache.flink.api.java.tuple;
 
-public class Tuple0 extends Tuple {
+import java.io.ObjectStreamException;
 
+/**
+ * A tuple with 0 fields.
+ * <p>
+ * {@code Tuple0} is a singleton.
+ * 
+ * @see Tuple
+ */
+public class Tuple0 extends Tuple {
 	private static final long serialVersionUID = 1L;
 
-	public Tuple0() {}
+	public static final Tuple0 instance = new Tuple0();
+
+	@Override
+	public int getArity() {
+		return 0;
+	}
 
 	@Override
 	public <T> T getField(int pos) {
-		return null;
+		throw new IndexOutOfBoundsException(String.valueOf(pos));
 	}
 
 	@Override
 	public <T> void setField(T value, int pos) {
+		throw new IndexOutOfBoundsException(String.valueOf(pos));
+	}
+
+	// -------------------------------------------------------------------------------------------------
+	// standard utilities
+	// -------------------------------------------------------------------------------------------------
+
+	/**
+	 * Creates a string representation of the tuple in the form "()"
+	 * 
+	 * @return The string representation of the tuple.
+	 */
+	@Override
+	public String toString() {
+		return "()";
 	}
 
+	/**
+	 * Deep equality for tuples by calling equals() on the tuple members
+	 * 
+	 * @param o
+	 *            the object checked for equality
+	 * @return true if this is equal to o.
+	 */
 	@Override
-	public int getArity() {
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (!(o instanceof Tuple0)) {
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	public int hashCode() {
 		return 0;
 	}
+
+	// singleton deserialization
+	private Object readResolve() throws ObjectStreamException {
+		return instance;
+	}
+
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
index a87fa0a..6ab02e4 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/TupleGenerator.java
@@ -402,12 +402,9 @@ class TupleGenerator {
 	private static void modifyTupleType(File root) throws IOException {
 		// generate code
 		StringBuilder sb = new StringBuilder();
-		sb.append("\tprivate static final Class<?>[] CLASSES = new Class<?>[] {\n\t\t");
+		sb.append("\tprivate static final Class<?>[] CLASSES = new Class<?>[] {\n\t\tTuple0.class");
 		for (int i = FIRST; i <= LAST; i++) {
-			if (i > FIRST) {
-				sb.append(", ");
-			}
-			sb.append("Tuple" + i + ".class");
+			sb.append(", Tuple" + i + ".class");
 		}
 		sb.append("\n\t};");
 
@@ -802,7 +799,7 @@ class TupleGenerator {
 		// package and imports
 		w.println("package " + PACKAGE + "." + BUILDER_SUFFIX + ';');
 		w.println();
-		w.println("import java.util.LinkedList;");
+		w.println("import java.util.ArrayList;");
 		w.println("import java.util.List;");
 		w.println();
 		w.println("import " + PACKAGE + ".Tuple" + numFields + ";");
@@ -817,7 +814,7 @@ class TupleGenerator {
 		// Class-Attributes - a list of tuples
 		w.print("\tprivate List<Tuple" + numFields);
 		printGenericsString(w, numFields);
-		w.print("> tuples = new LinkedList<Tuple" + numFields );
+		w.print("> tuples = new ArrayList<Tuple" + numFields );
 		printGenericsString(w, numFields);
 		w.println(">();");
 		w.println();

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple0Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple0Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple0Builder.java
new file mode 100644
index 0000000..2a1546e
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple0Builder.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+// --------------------------------------------------------------
+//  THIS IS A GENERATED SOURCE FILE. DO NOT EDIT!
+//  GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
+// --------------------------------------------------------------
+
+
+package org.apache.flink.api.java.tuple.builder;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple0;
+
+public class Tuple0Builder {
+
+	private List<Tuple0> tuples = new ArrayList<Tuple0>();
+
+	public Tuple0Builder add() {
+		tuples.add(Tuple0.instance);
+		return this;
+	}
+
+	public Tuple0[] build() {
+		return tuples.toArray(new Tuple0[tuples.size()]);
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple10Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple10Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple10Builder.java
index 79e30a5..d09ba11 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple10Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple10Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple10;
 
 public class Tuple10Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9> {
 
-	private List<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> tuples = new LinkedList<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>>();
+	private List<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>> tuples = new ArrayList<Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>>();
 
 	public Tuple10Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9){
 		tuples.add(new Tuple10<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple11Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple11Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple11Builder.java
index d1a733a..727fab0 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple11Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple11Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple11;
 
 public class Tuple11Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10> {
 
-	private List<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> tuples = new LinkedList<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>>();
+	private List<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>> tuples = new ArrayList<Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>>();
 
 	public Tuple11Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10){
 		tuples.add(new Tuple11<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple12Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple12Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple12Builder.java
index bbc8967..7f77a8d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple12Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple12Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple12;
 
 public class Tuple12Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11> {
 
-	private List<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> tuples = new LinkedList<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>>();
+	private List<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>> tuples = new ArrayList<Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>>();
 
 	public Tuple12Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11){
 		tuples.add(new Tuple12<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple13Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple13Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple13Builder.java
index 7f22f58..e223e7a 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple13Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple13Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple13;
 
 public class Tuple13Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12> {
 
-	private List<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> tuples = new LinkedList<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>>();
+	private List<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>> tuples = new ArrayList<Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>>();
 
 	public Tuple13Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12){
 		tuples.add(new Tuple13<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple14Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple14Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple14Builder.java
index a587f45..a3847c2 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple14Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple14Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple14;
 
 public class Tuple14Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13> {
 
-	private List<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> tuples = new LinkedList<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>>();
+	private List<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>> tuples = new ArrayList<Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>>();
 
 	public Tuple14Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13){
 		tuples.add(new Tuple14<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple15Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple15Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple15Builder.java
index 8068389..b423a3b 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple15Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple15Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple15;
 
 public class Tuple15Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14> {
 
-	private List<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> tuples = new LinkedList<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>>();
+	private List<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>> tuples = new ArrayList<Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>>();
 
 	public Tuple15Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14){
 		tuples.add(new Tuple15<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple16Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple16Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple16Builder.java
index 478c36d..c698730 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple16Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple16Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple16;
 
 public class Tuple16Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15> {
 
-	private List<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> tuples = new LinkedList<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>>();
+	private List<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>> tuples = new ArrayList<Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>>();
 
 	public Tuple16Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15){
 		tuples.add(new Tuple16<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple17Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple17Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple17Builder.java
index 764b3eb..bad64f2 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple17Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple17Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple17;
 
 public class Tuple17Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16> {
 
-	private List<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> tuples = new LinkedList<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>>();
+	private List<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>> tuples = new ArrayList<Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>>();
 
 	public Tuple17Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16){
 		tuples.add(new Tuple17<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple18Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple18Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple18Builder.java
index 6a830e2..14a79f8 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple18Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple18Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple18;
 
 public class Tuple18Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17> {
 
-	private List<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> tuples = new LinkedList<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>>();
+	private List<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>> tuples = new ArrayList<Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>>();
 
 	public Tuple18Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17){
 		tuples.add(new Tuple18<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple19Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple19Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple19Builder.java
index 08d14cc..9acfc2f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple19Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple19Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple19;
 
 public class Tuple19Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18> {
 
-	private List<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> tuples = new LinkedList<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>>();
+	private List<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>> tuples = new ArrayList<Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>>();
 
 	public Tuple19Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18){
 		tuples.add(new Tuple19<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple1Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple1Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple1Builder.java
index a8c2ce2..a395cd1 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple1Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple1Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple1;
 
 public class Tuple1Builder<T0> {
 
-	private List<Tuple1<T0>> tuples = new LinkedList<Tuple1<T0>>();
+	private List<Tuple1<T0>> tuples = new ArrayList<Tuple1<T0>>();
 
 	public Tuple1Builder<T0> add(T0 value0){
 		tuples.add(new Tuple1<T0>(value0));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple20Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple20Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple20Builder.java
index 139292e..8c7dd1d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple20Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple20Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple20;
 
 public class Tuple20Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19> {
 
-	private List<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> tuples = new LinkedList<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>>();
+	private List<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>> tuples = new ArrayList<Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>>();
 
 	public Tuple20Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19){
 		tuples.add(new Tuple20<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple21Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple21Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple21Builder.java
index e6481f8..ba18345 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple21Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple21Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple21;
 
 public class Tuple21Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20> {
 
-	private List<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> tuples = new LinkedList<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>>();
+	private List<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>> tuples = new ArrayList<Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>>();
 
 	public Tuple21Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19, T20 value20){
 		tuples.add(new Tuple21<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple22Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple22Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple22Builder.java
index 0f4f939..a4903c8 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple22Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple22Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple22;
 
 public class Tuple22Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21> {
 
-	private List<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> tuples = new LinkedList<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>>();
+	private List<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>> tuples = new ArrayList<Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>>();
 
 	public Tuple22Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19, T20 value20, T21 value21){
 		tuples.add(new Tuple22<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20, value21));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple23Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple23Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple23Builder.java
index 66e0313..a7a8102 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple23Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple23Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple23;
 
 public class Tuple23Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22> {
 
-	private List<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> tuples = new LinkedList<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>>();
+	private List<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>> tuples = new ArrayList<Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>>();
 
 	public Tuple23Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19, T20 value20, T21 value21, T22 value22){
 		tuples.add(new Tuple23<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20, value21, value22));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple24Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple24Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple24Builder.java
index 9aac48e..6509518 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple24Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple24Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple24;
 
 public class Tuple24Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23> {
 
-	private List<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> tuples = new LinkedList<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>>();
+	private List<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>> tuples = new ArrayList<Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>>();
 
 	public Tuple24Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19, T20 value20, T21 value21, T22 value22, T23 value23){
 		tuples.add(new Tuple24<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20, value21, value22, value23));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple25Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple25Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple25Builder.java
index 087b8cc..3632eec 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple25Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple25Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple25;
 
 public class Tuple25Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24> {
 
-	private List<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> tuples = new LinkedList<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>>();
+	private List<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>> tuples = new ArrayList<Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>>();
 
 	public Tuple25Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8, T9 value9, T10 value10, T11 value11, T12 value12, T13 value13, T14 value14, T15 value15, T16 value16, T17 value17, T18 value18, T19 value19, T20 value20, T21 value21, T22 value22, T23 value23, T24 value24){
 		tuples.add(new Tuple25<T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16, T17, T18, T19, T20, T21, T22, T23, T24>(value0, value1, value2, value3, value4, value5, value6, value7, value8, value9, value10, value11, value12, value13, value14, value15, value16, value17, value18, value19, value20, value21, value22, value23, value24));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple2Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple2Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple2Builder.java
index b53327d..adf697f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple2Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple2Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple2;
 
 public class Tuple2Builder<T0, T1> {
 
-	private List<Tuple2<T0, T1>> tuples = new LinkedList<Tuple2<T0, T1>>();
+	private List<Tuple2<T0, T1>> tuples = new ArrayList<Tuple2<T0, T1>>();
 
 	public Tuple2Builder<T0, T1> add(T0 value0, T1 value1){
 		tuples.add(new Tuple2<T0, T1>(value0, value1));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple3Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple3Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple3Builder.java
index e6948e9..7a0dee0 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple3Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple3Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple3;
 
 public class Tuple3Builder<T0, T1, T2> {
 
-	private List<Tuple3<T0, T1, T2>> tuples = new LinkedList<Tuple3<T0, T1, T2>>();
+	private List<Tuple3<T0, T1, T2>> tuples = new ArrayList<Tuple3<T0, T1, T2>>();
 
 	public Tuple3Builder<T0, T1, T2> add(T0 value0, T1 value1, T2 value2){
 		tuples.add(new Tuple3<T0, T1, T2>(value0, value1, value2));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple4Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple4Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple4Builder.java
index cdb79c6..5caad15 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple4Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple4Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple4;
 
 public class Tuple4Builder<T0, T1, T2, T3> {
 
-	private List<Tuple4<T0, T1, T2, T3>> tuples = new LinkedList<Tuple4<T0, T1, T2, T3>>();
+	private List<Tuple4<T0, T1, T2, T3>> tuples = new ArrayList<Tuple4<T0, T1, T2, T3>>();
 
 	public Tuple4Builder<T0, T1, T2, T3> add(T0 value0, T1 value1, T2 value2, T3 value3){
 		tuples.add(new Tuple4<T0, T1, T2, T3>(value0, value1, value2, value3));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple5Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple5Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple5Builder.java
index c1b21a2..1895aca 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple5Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple5Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple5;
 
 public class Tuple5Builder<T0, T1, T2, T3, T4> {
 
-	private List<Tuple5<T0, T1, T2, T3, T4>> tuples = new LinkedList<Tuple5<T0, T1, T2, T3, T4>>();
+	private List<Tuple5<T0, T1, T2, T3, T4>> tuples = new ArrayList<Tuple5<T0, T1, T2, T3, T4>>();
 
 	public Tuple5Builder<T0, T1, T2, T3, T4> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4){
 		tuples.add(new Tuple5<T0, T1, T2, T3, T4>(value0, value1, value2, value3, value4));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple6Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple6Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple6Builder.java
index 9334be1..625ec8f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple6Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple6Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple6;
 
 public class Tuple6Builder<T0, T1, T2, T3, T4, T5> {
 
-	private List<Tuple6<T0, T1, T2, T3, T4, T5>> tuples = new LinkedList<Tuple6<T0, T1, T2, T3, T4, T5>>();
+	private List<Tuple6<T0, T1, T2, T3, T4, T5>> tuples = new ArrayList<Tuple6<T0, T1, T2, T3, T4, T5>>();
 
 	public Tuple6Builder<T0, T1, T2, T3, T4, T5> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5){
 		tuples.add(new Tuple6<T0, T1, T2, T3, T4, T5>(value0, value1, value2, value3, value4, value5));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple7Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple7Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple7Builder.java
index ebc2166..2cdde39 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple7Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple7Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple7;
 
 public class Tuple7Builder<T0, T1, T2, T3, T4, T5, T6> {
 
-	private List<Tuple7<T0, T1, T2, T3, T4, T5, T6>> tuples = new LinkedList<Tuple7<T0, T1, T2, T3, T4, T5, T6>>();
+	private List<Tuple7<T0, T1, T2, T3, T4, T5, T6>> tuples = new ArrayList<Tuple7<T0, T1, T2, T3, T4, T5, T6>>();
 
 	public Tuple7Builder<T0, T1, T2, T3, T4, T5, T6> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6){
 		tuples.add(new Tuple7<T0, T1, T2, T3, T4, T5, T6>(value0, value1, value2, value3, value4, value5, value6));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple8Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple8Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple8Builder.java
index dd0860f..c5feb43 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple8Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple8Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple8;
 
 public class Tuple8Builder<T0, T1, T2, T3, T4, T5, T6, T7> {
 
-	private List<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> tuples = new LinkedList<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>>();
+	private List<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>> tuples = new ArrayList<Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>>();
 
 	public Tuple8Builder<T0, T1, T2, T3, T4, T5, T6, T7> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7){
 		tuples.add(new Tuple8<T0, T1, T2, T3, T4, T5, T6, T7>(value0, value1, value2, value3, value4, value5, value6, value7));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple9Builder.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple9Builder.java b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple9Builder.java
index c663985..2665d7d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple9Builder.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/tuple/builder/Tuple9Builder.java
@@ -25,14 +25,14 @@
 
 package org.apache.flink.api.java.tuple.builder;
 
-import java.util.LinkedList;
+import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.flink.api.java.tuple.Tuple9;
 
 public class Tuple9Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8> {
 
-	private List<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> tuples = new LinkedList<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>>();
+	private List<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>> tuples = new ArrayList<Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>>();
 
 	public Tuple9Builder<T0, T1, T2, T3, T4, T5, T6, T7, T8> add(T0 value0, T1 value1, T2 value2, T3 value3, T4 value4, T5 value5, T6 value6, T7 value7, T8 value8){
 		tuples.add(new Tuple9<T0, T1, T2, T3, T4, T5, T6, T7, T8>(value0, value1, value2, value3, value4, value5, value6, value7, value8));

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
index 0e3d61d..0e59b8e 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeComparator;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.*;
+import org.apache.flink.api.java.typeutils.runtime.Tuple0Serializer;
 //CHECKSTYLE.ON: AvoidStarImport
 import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
 import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
@@ -49,7 +50,7 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> {
 
 	public TupleTypeInfo(Class<T> tupleType, TypeInformation<?>... types) {
 		super(tupleType, types);
-		if (types == null || types.length == 0 || types.length > Tuple.MAX_ARITY) {
+		if (types == null || types.length > Tuple.MAX_ARITY) {
 			throw new IllegalArgumentException();
 		}
 		this.fieldNames = new String[types.length];
@@ -72,8 +73,13 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> {
 		return fieldIndex;
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public TupleSerializer<T> createSerializer(ExecutionConfig executionConfig) {
+		if (this.tupleType == Tuple0.class) {
+			return (TupleSerializer<T>) Tuple0Serializer.getInstance();
+		}
+
 		TypeSerializer<?>[] fieldSerializers = new TypeSerializer<?>[getArity()];
 		for (int i = 0; i < types.length; i++) {
 			fieldSerializers[i] = types[i].createSerializer(executionConfig);

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
index 881e690..a2d937f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
@@ -160,6 +160,7 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 		}
 	}
 
+	@Override
 	public <X> TypeInformation<X> getTypeAt(String fieldExpression) {
 
 		Matcher matcher = PATTERN_NESTED_FIELDS.matcher(fieldExpression);
@@ -196,6 +197,7 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 		}
 	}
 	
+	@Override
 	public <X> TypeInformation<X> getTypeAt(int pos) {
 		if (pos < 0 || pos >= this.types.length) {
 			throw new IndexOutOfBoundsException();
@@ -227,14 +229,16 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 	@Override
 	public String toString() {
 		StringBuilder bld = new StringBuilder("Tuple");
-		bld.append(types.length).append('<');
-		bld.append(types[0]);
-		
-		for (int i = 1; i < types.length; i++) {
-			bld.append(", ").append(types[i]);
+		bld.append(types.length);
+		if (types.length > 0) {
+			bld.append('<').append(types[0]);
+
+			for (int i = 1; i < types.length; i++) {
+				bld.append(", ").append(types[i]);
+			}
+
+			bld.append('>');
 		}
-		
-		bld.append('>');
 		return bld.toString();
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
index 2e45107..8f5d599 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java
@@ -52,6 +52,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple0;
 import org.apache.flink.types.Value;
 import org.apache.flink.util.Collector;
 import org.apache.hadoop.io.Writable;
@@ -408,6 +409,10 @@ public class TypeExtractor {
 				curT = typeToClass(curT).getGenericSuperclass();
 			}
 			
+			if(curT == Tuple0.class) {
+				return new TupleTypeInfo(Tuple0.class, new TypeInformation<?>[0]);
+			}
+			
 			// check if immediate child of Tuple has generics
 			if (curT instanceof Class<?>) {
 				throw new InvalidTypesException("Tuple needs to be parameterized by using generics.");
@@ -684,7 +689,7 @@ public class TypeExtractor {
 	
 	private static Type getParameterTypeFromGenericType(Class<?> baseClass, ArrayList<Type> typeHierarchy, Type t, int pos) {
 		// base class
-		if (t instanceof ParameterizedType && baseClass.equals((Class<?>) ((ParameterizedType) t).getRawType())) {
+		if (t instanceof ParameterizedType && baseClass.equals(((ParameterizedType) t).getRawType())) {
 			if (typeHierarchy != null) {
 				typeHierarchy.add(t);
 			}
@@ -784,6 +789,10 @@ public class TypeExtractor {
 					type = typeToClass(type).getGenericSuperclass();
 				}
 				
+				if(type == Tuple0.class) {
+					return;
+				}
+				
 				// check if immediate child of Tuple has generics
 				if (type instanceof Class<?>) {
 					throw new InvalidTypesException("Parameterized Tuple type expected.");
@@ -1202,17 +1211,20 @@ public class TypeExtractor {
 		
 		// check for subclasses of Tuple
 		if (Tuple.class.isAssignableFrom(clazz)) {
-			throw new InvalidTypesException("Type information extraction for tuples cannot be done based on the class.");
+			if(clazz == Tuple0.class) {
+				return new TupleTypeInfo(Tuple0.class, new TypeInformation<?>[0]);
+			}
+			throw new InvalidTypesException("Type information extraction for tuples (except Tuple0) cannot be done based on the class.");
 		}
 
 		// check for Enums
 		if(Enum.class.isAssignableFrom(clazz)) {
-			return (TypeInformation<OUT>) new EnumTypeInfo(clazz);
+			return new EnumTypeInfo(clazz);
 		}
 
 		// special case for POJOs generated by Avro.
 		if(SpecificRecordBase.class.isAssignableFrom(clazz)) {
-			return (TypeInformation<OUT>) new AvroTypeInfo(clazz);
+			return new AvroTypeInfo(clazz);
 		}
 
 		if (countTypeInHierarchy(typeHierarchy, clazz) > 1) {
@@ -1506,7 +1518,7 @@ public class TypeExtractor {
 				
 				infos[i] = privateGetForObject(field);
 			}
-			return (TypeInformation<X>) new TupleTypeInfo(value.getClass(), infos);
+			return new TupleTypeInfo(value.getClass(), infos);
 		} else {
 			return privateGetForClass((Class<X>) value.getClass(), new ArrayList<Type>());
 		}

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java
index 2e04aca..b89a830 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TypeInfoParser.java
@@ -36,7 +36,7 @@ public class TypeInfoParser {
 	private static final String VALUE_PACKAGE = "org.apache.flink.types";
 	private static final String WRITABLE_PACKAGE = "org.apache.hadoop.io";
 
-	private static final Pattern tuplePattern = Pattern.compile("^((" + TUPLE_PACKAGE.replaceAll("\\.", "\\\\.") + "\\.)?Tuple[0-9]+)<");
+	private static final Pattern tuplePattern = Pattern.compile("^(" + TUPLE_PACKAGE.replaceAll("\\.", "\\\\.") + "\\.)?((Tuple[1-9][0-9]?)<|(Tuple0))");
 	private static final Pattern writablePattern = Pattern.compile("^((" + WRITABLE_PACKAGE.replaceAll("\\.", "\\\\.") + "\\.)?Writable)<([^\\s,>]*)(,|>|$|\\[)");
 	private static final Pattern enumPattern = Pattern.compile("^((java\\.lang\\.)?Enum)<([^\\s,>]*)(,|>|$|\\[)");
 	private static final Pattern basicTypePattern = Pattern
@@ -124,18 +124,23 @@ public class TypeInfoParser {
 
 		// tuples
 		if (tupleMatcher.find()) {
-			String className = tupleMatcher.group(1);
-			sb.delete(0, className.length() + 1);
-			int arity = Integer.parseInt(className.replaceAll("\\D", ""));
-
-			Class<?> clazz;
-			// check if fully qualified
-			if (className.startsWith(TUPLE_PACKAGE)) {
-				clazz = loadClass(className);
+			boolean isGenericTuple = true;
+			String className = tupleMatcher.group(3);
+			if(className == null) { // matched Tuple0
+				isGenericTuple = false;
+				className = tupleMatcher.group(2);
+				sb.delete(0, className.length());
 			} else {
-				clazz = loadClass(TUPLE_PACKAGE + "." + className);
+				sb.delete(0, className.length() + 1); // +1 for "<"
+			}
+
+			if (infoString.startsWith(TUPLE_PACKAGE)) {
+				sb.delete(0, TUPLE_PACKAGE.length() + 1); // +1 for trailing "."
 			}
 
+			int arity = Integer.parseInt(className.replaceAll("\\D", ""));
+			Class<?> clazz = loadClass(TUPLE_PACKAGE + "." + className);
+
 			TypeInformation<?>[] types = new TypeInformation<?>[arity];
 			for (int i = 0; i < arity; i++) {
 				types[i] = parse(sb);
@@ -143,11 +148,13 @@ public class TypeInfoParser {
 					throw new IllegalArgumentException("Tuple arity does not match given parameters.");
 				}
 			}
-			if (sb.charAt(0) != '>') {
-				throw new IllegalArgumentException("Tuple arity does not match given parameters.");
+			if (isGenericTuple) {
+				if(sb.charAt(0) != '>') {
+					throw new IllegalArgumentException("Tuple arity does not match given parameters.");
+				}
+				// remove '>'
+				sb.deleteCharAt(0);
 			}
-			// remove '>'
-			sb.deleteCharAt(0);
 			returnType = new TupleTypeInfo(clazz, types);
 		}
 		// writable types

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
new file mode 100644
index 0000000..e2d43e1
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/Tuple0Serializer.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
+ * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.apache.flink.api.java.typeutils.runtime;
+
+import java.io.IOException;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple0;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+
+public class Tuple0Serializer extends TupleSerializer<Tuple0> {
+	private static final long serialVersionUID = 1278813169022975971L;
+
+	private static final Tuple0Serializer singleton = new Tuple0Serializer();
+
+	private Tuple0Serializer() {
+		super(Tuple0.class, new TypeSerializer<?>[0]);
+	}
+
+	public static Tuple0Serializer getInstance() {
+		return singleton;
+	}
+
+	@Override
+	public Tuple0Serializer duplicate() {
+		return this;
+	}
+
+	@Override
+	public Tuple0 createInstance() {
+		return Tuple0.instance;
+	}
+
+	@Override
+	public Tuple0 createInstance(Object[] fields) {
+		if (fields == null || fields.length == 0) {
+			return Tuple0.instance;
+		}
+
+		throw new UnsupportedOperationException(
+				"Tuple0 cannot take any data, as it has zero fields.");
+	}
+
+	@Override
+	public Tuple0 copy(Tuple0 from) {
+		return Tuple0.instance;
+	}
+
+	@Override
+	public Tuple0 copy(Tuple0 from, Tuple0 reuse) {
+		return reuse;
+	}
+
+	@Override
+	public int getLength() {
+		return 1;
+	}
+
+	@Override
+	public void serialize(Tuple0 record, DataOutputView target) throws IOException {
+		target.writeByte(42);
+	}
+
+	@Override
+	public Tuple0 deserialize(DataInputView source) throws IOException {
+		source.readByte();
+		return Tuple0.instance;
+	}
+
+	@Override
+	public Tuple0 deserialize(Tuple0 reuse, DataInputView source) throws IOException {
+		source.readByte();
+		return reuse;
+	}
+
+	@Override
+	public void copy(DataInputView source, DataOutputView target) throws IOException {
+		target.writeByte(source.readByte());
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
index 2b330c2..46e3990 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
@@ -27,7 +27,7 @@ import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.types.NullFieldException;
 
 
-public final class TupleSerializer<T extends Tuple> extends TupleSerializerBase<T> {
+public class TupleSerializer<T extends Tuple> extends TupleSerializerBase<T> {
 
 	private static final long serialVersionUID = 1L;
 	

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java
index 925e8e5..d27a82b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/type/extractor/TypeExtractorTest.java
@@ -40,6 +40,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple0;
 import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple3;
@@ -267,6 +268,27 @@ public class TypeExtractorTest {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
+	public void testTuple0() {
+		// use getFlatMapReturnTypes()
+		RichFlatMapFunction<?, ?> function = new RichFlatMapFunction<Tuple0, Tuple0>() {
+			private static final long serialVersionUID = 1L;
+
+			@Override
+			public void flatMap(Tuple0 value, Collector<Tuple0> out) throws Exception {
+				// nothing to do
+			}
+		};
+
+		TypeInformation<?> ti = TypeExtractor.getFlatMapReturnTypes(function,
+				(TypeInformation) TypeInfoParser.parse("Tuple0"));
+
+		Assert.assertTrue(ti.isTupleType());
+		Assert.assertEquals(0, ti.getArity());
+		Assert.assertTrue(ti instanceof TupleTypeInfo);
+	}
+	
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
 	public void testSubclassOfTuple() {
 		// use getJoinReturnTypes()
 		RichFlatJoinFunction<?, ?, ?> function = new RichFlatJoinFunction<CustomTuple, String, CustomTuple>() {

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoParserTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoParserTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoParserTest.java
index eadf96d..9fe8174 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoParserTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/TypeInfoParserTest.java
@@ -144,6 +144,14 @@ public class TypeInfoParserTest {
 		Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ((TupleTypeInfo<?>)ti).getTypeAt(0));
 		Assert.assertEquals(BasicTypeInfo.LONG_TYPE_INFO, ((TupleTypeInfo<?>)ti).getTypeAt(1));
 		
+		ti = TypeInfoParser.parse("Tuple0");
+		Assert.assertEquals(0, ti.getArity());
+		Assert.assertEquals("Java Tuple0", ti.toString());
+		
+		ti = TypeInfoParser.parse("org.apache.flink.api.java.tuple.Tuple0");
+		Assert.assertEquals(0, ti.getArity());
+		Assert.assertEquals("Java Tuple0", ti.toString());
+		
 		ti = TypeInfoParser.parse("Tuple3<Tuple1<String>, Tuple1<Integer>, Tuple2<Long, Long>>");
 		Assert.assertEquals("Java Tuple3<Java Tuple1<String>, Java Tuple1<Integer>, Java Tuple2<Long, Long>>", ti.toString());
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerTest.java
index 96f8306..beda8e9 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializerTest.java
@@ -25,6 +25,7 @@ import java.util.Random;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple0;
 import org.apache.flink.api.java.tuple.Tuple1;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple5;
@@ -40,7 +41,14 @@ import org.junit.Assert;
 import org.junit.Test;
 
 public class TupleSerializerTest {
-	
+
+	@Test
+	public void testTuple0() {
+		Tuple0[] testTuples = new Tuple0[] { Tuple0.instance, Tuple0.instance, Tuple0.instance };
+
+		runTests(testTuples);
+	}
+
 	@Test
 	public void testTuple1Int() {
 		@SuppressWarnings("unchecked")
@@ -214,7 +222,11 @@ public class TupleSerializerTest {
 			
 			Class<T> tupleClass = tupleTypeInfo.getTypeClass();
 			
-			TupleSerializerTestInstance<T> test = new TupleSerializerTestInstance<T>(serializer, tupleClass, -1, instances);
+			int length = -1;
+			if(tupleClass == Tuple0.class) {
+				length = 1;
+			}
+			TupleSerializerTestInstance<T> test = new TupleSerializerTestInstance<T>(serializer, tupleClass, length, instances);
 			test.testAll();
 		}
 		catch (Exception e) {

http://git-wip-us.apache.org/repos/asf/flink/blob/fab9ce5d/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
index 23720d7..59ed20c 100644
--- a/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
+++ b/flink-staging/flink-language-binding/flink-language-binding-generic/src/main/java/org/apache/flink/languagebinding/api/java/common/streaming/Receiver.java
@@ -32,6 +32,8 @@ import org.apache.flink.util.Collector;
  * General-purpose class to read data from memory-mapped files.
  */
 public class Receiver implements Serializable {
+	private static final long serialVersionUID = -2474088929850009968L;
+
 	private final AbstractRichFunction function;
 
 	private File inputFile;
@@ -39,7 +41,7 @@ public class Receiver implements Serializable {
 	private FileChannel inputChannel;
 	private MappedByteBuffer fileBuffer;
 
-	private Deserializer deserializer = null;
+	private Deserializer<?> deserializer = null;
 
 	public Receiver(AbstractRichFunction function) {
 		this.function = function;
@@ -196,6 +198,7 @@ public class Receiver implements Serializable {
 	 * @param bufferSize size of the buffer
 	 * @throws IOException
 	 */
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void collectBuffer(Collector c, int bufferSize) throws IOException {
 		fileBuffer.position(0);
 
@@ -209,7 +212,7 @@ public class Receiver implements Serializable {
 	}
 
 	//=====Deserializer=================================================================================================
-	private Deserializer getDeserializer(byte type) {
+	private Deserializer<?> getDeserializer(byte type) {
 		switch (type) {
 			case TYPE_TUPLE:
 				return new TupleDeserializer();
@@ -324,7 +327,7 @@ public class Receiver implements Serializable {
 	}
 
 	private class TupleDeserializer implements Deserializer<Tuple> {
-		Deserializer[] deserializer = null;
+		Deserializer<?>[] deserializer = null;
 		Tuple reuse;
 
 		public TupleDeserializer() {
@@ -346,61 +349,12 @@ public class Receiver implements Serializable {
 	}
 
 	public static Tuple createTuple(int size) {
-		switch (size) {
-			case 0:
-				return new Tuple0();
-			case 1:
-				return new Tuple1();
-			case 2:
-				return new Tuple2();
-			case 3:
-				return new Tuple3();
-			case 4:
-				return new Tuple4();
-			case 5:
-				return new Tuple5();
-			case 6:
-				return new Tuple6();
-			case 7:
-				return new Tuple7();
-			case 8:
-				return new Tuple8();
-			case 9:
-				return new Tuple9();
-			case 10:
-				return new Tuple10();
-			case 11:
-				return new Tuple11();
-			case 12:
-				return new Tuple12();
-			case 13:
-				return new Tuple13();
-			case 14:
-				return new Tuple14();
-			case 15:
-				return new Tuple15();
-			case 16:
-				return new Tuple16();
-			case 17:
-				return new Tuple17();
-			case 18:
-				return new Tuple18();
-			case 19:
-				return new Tuple19();
-			case 20:
-				return new Tuple20();
-			case 21:
-				return new Tuple21();
-			case 22:
-				return new Tuple22();
-			case 23:
-				return new Tuple23();
-			case 24:
-				return new Tuple24();
-			case 25:
-				return new Tuple25();
-			default:
-				throw new IllegalArgumentException("Tuple size not supported: " + size);
+		try {
+			return Tuple.getTupleClass(size).newInstance();
+		} catch (InstantiationException e) {
+			throw new RuntimeException(e);
+		} catch (IllegalAccessException e) {
+			throw new RuntimeException(e);
 		}
 	}
 }


Mime
View raw message