You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/02/16 20:40:29 UTC
[3/3] flink git commit: [core] [java api] Various code cleanup and
fixed for warnings.
[core] [java api] Various code cleanup and fixed for warnings.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6ba5fcdf
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6ba5fcdf
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6ba5fcdf
Branch: refs/heads/master
Commit: 6ba5fcdf85f802748fd47218ce8c41a617d224c9
Parents: 2dcff4c
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Feb 16 15:06:49 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Feb 16 20:39:52 2015 +0100
----------------------------------------------------------------------
.../api/common/InvalidProgramException.java | 7 +-
.../java/org/apache/flink/api/common/Plan.java | 1 -
.../api/common/io/GenericCsvInputFormat.java | 8 +-
.../common/operators/GenericDataSinkBase.java | 3 +-
.../flink/api/common/typeinfo/AtomicType.java | 17 +++-
.../api/common/typeinfo/BasicArrayTypeInfo.java | 2 +
.../api/common/typeinfo/BasicTypeInfo.java | 2 +
.../api/common/typeinfo/NothingTypeInfo.java | 7 +-
.../common/typeinfo/PrimitiveArrayTypeInfo.java | 32 +++++++
.../api/common/typeinfo/TypeInformation.java | 99 ++++++++++++++++++--
.../api/common/typeutils/CompositeType.java | 4 +
.../typeutils/CompositeTypeComparator.java | 1 +
.../apache/flink/types/parser/StringParser.java | 2 -
.../flink/types/parser/StringValueParser.java | 2 -
.../apache/flink/util/SplittableIterator.java | 2 +
.../java/hadoop/mapred/HadoopInputFormat.java | 3 +-
.../java/hadoop/mapred/HadoopOutputFormat.java | 3 +-
.../hadoop/mapred/wrapper/HadoopInputSplit.java | 9 +-
.../hadoop/mapreduce/HadoopInputFormat.java | 3 +-
.../hadoop/mapreduce/HadoopOutputFormat.java | 3 +-
.../mapreduce/HadoopOutputFormatBase.java | 1 -
.../mapreduce/wrapper/HadoopInputSplit.java | 10 +-
.../flink/api/java/typeutils/EnumTypeInfo.java | 11 +++
.../api/java/typeutils/GenericTypeInfo.java | 2 +
.../api/java/typeutils/MissingTypeInfo.java | 2 +
.../api/java/typeutils/ObjectArrayTypeInfo.java | 2 +
.../flink/api/java/typeutils/PojoTypeInfo.java | 13 ++-
.../api/java/typeutils/RecordTypeInfo.java | 5 +-
.../flink/api/java/typeutils/TupleTypeInfo.java | 10 +-
.../api/java/typeutils/TupleTypeInfoBase.java | 4 +
.../flink/api/java/typeutils/ValueTypeInfo.java | 9 +-
.../api/java/typeutils/WritableTypeInfo.java | 8 ++
.../java/typeutils/runtime/AvroSerializer.java | 4 +-
.../java/typeutils/runtime/PojoSerializer.java | 24 ++---
.../java/typeutils/runtime/TupleSerializer.java | 2 +-
.../typeutils/runtime/WritableSerializer.java | 4 +-
.../hadoop/mapred/HadoopInputFormatTest.java | 20 +---
.../hadoop/mapreduce/HadoopInputFormatTest.java | 19 +---
.../apache/flink/runtime/client/JobClient.scala | 3 +-
.../flink/runtime/taskmanager/TaskManager.scala | 3 +-
.../jobmanager/JobManagerStartupTest.java | 4 +-
41 files changed, 277 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/InvalidProgramException.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/InvalidProgramException.java b/flink-core/src/main/java/org/apache/flink/api/common/InvalidProgramException.java
index 6753982..e707709 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/InvalidProgramException.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/InvalidProgramException.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common;
/**
@@ -44,6 +43,12 @@ public class InvalidProgramException extends RuntimeException {
super(message);
}
+ /**
+ * Creates a new exception with the given message and cause.
+ *
+ * @param message The exception message.
+ * @param e The exception cause.
+ */
public InvalidProgramException(String message, Throwable e) {
super(message, e);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java b/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
index fb50742..88e9c87 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/Plan.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common;
import static com.google.common.base.Preconditions.checkArgument;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
index 662cf1b..31a2a5a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
@@ -22,6 +22,7 @@ package org.apache.flink.api.common.io;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.primitives.Ints;
+
import org.apache.flink.core.fs.FileInputSplit;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.parser.FieldParser;
@@ -261,7 +262,6 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
super.open(split);
// instantiate the parsers
- @SuppressWarnings("unchecked")
FieldParser<?>[] parsers = new FieldParser[fieldTypes.length];
for (int i = 0; i < fieldTypes.length; i++) {
@@ -270,10 +270,8 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
if (parserType == null) {
throw new RuntimeException("No parser available for type '" + fieldTypes[i].getName() + "'.");
}
-
- @SuppressWarnings("unchecked")
- FieldParser<?> p = (FieldParser<?>) InstantiationUtil.instantiate(parserType, FieldParser.class);
+ FieldParser<?> p = InstantiationUtil.instantiate(parserType, FieldParser.class);
if (this.quotedStringParsing) {
if (p instanceof StringParser) {
@@ -314,6 +312,7 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
if (fieldIncluded[field]) {
// parse field
+ @SuppressWarnings("unchecked")
FieldParser<Object> parser = (FieldParser<Object>) this.fieldParsers[output];
Object reuse = holders[output];
startPos = parser.parseField(bytes, startPos, limit, this.fieldDelim, reuse);
@@ -364,7 +363,6 @@ public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT>
protected int skipFields(byte[] bytes, int startPos, int limit, byte[] delim) {
int i = startPos;
- byte current;
final int delimLimit = limit - delim.length + 1;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
index 2f8a396..57f427a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/GenericDataSinkBase.java
@@ -299,6 +299,7 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
// --------------------------------------------------------------------------------------------
+ @SuppressWarnings("unchecked")
protected void executeOnCollections(List<IN> inputData, ExecutionConfig executionConfig) throws Exception {
OutputFormat<IN> format = this.formatWrapper.getUserCodeObject();
TypeInformation<IN> inputType = getInput().getOperatorInfo().getOutputType();
@@ -311,7 +312,7 @@ public class GenericDataSinkBase<IN> extends Operator<Nothing> {
if (inputType instanceof CompositeType) {
sortComparator = ((CompositeType<IN>) inputType).createComparator(sortColumns, sortOrderings, 0, executionConfig);
} else if (inputType instanceof AtomicType) {
- sortComparator = ((AtomicType) inputType).createComparator(sortOrderings[0], executionConfig);
+ sortComparator = ((AtomicType<IN>) inputType).createComparator(sortOrderings[0], executionConfig);
} else {
throw new UnsupportedOperationException("Local output sorting does not support type "+inputType+" yet.");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/AtomicType.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/AtomicType.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/AtomicType.java
index 2fc8a1b..35e749d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/AtomicType.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/AtomicType.java
@@ -21,11 +21,24 @@ package org.apache.flink.api.common.typeinfo;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeComparator;
-
/**
- *
+ * An atomic type is a type that is treated as one indivisible unit and where the entire type acts
+ * as a key. The atomic type defines the method to create a comparator for this type as a key.
+ * Example atomic types are the basic types (int, long, String, ...) and comparable custom classes.
+ * <p>
+ * In contrast to atomic types are composite types, where the type information is aware of the individual
+ * fields and individual fields may be used as a key.
*/
public interface AtomicType<T> {
+ /**
+ * Creates a comparator for this type.
+ *
+ * @param sortOrderAscending True, if the comparator should define the order to be ascending,
+ * false, if the comparator should define the order to be descending.
+ * @param executionConfig The config from which the comparator will be parameterized. Parameterization
+ * includes for example registration of class tags for frameworks like Kryo.
+ * @return A comparator for this type.
+ */
TypeComparator<T> createComparator(boolean sortOrderAscending, ExecutionConfig executionConfig);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
index 80f5f63..4fcb266 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
@@ -29,6 +29,8 @@ import org.apache.flink.api.common.typeutils.base.GenericArraySerializer;
public class BasicArrayTypeInfo<T, C> extends TypeInformation<T> {
+ private static final long serialVersionUID = 1L;
+
public static final BasicArrayTypeInfo<String[], String> STRING_ARRAY_TYPE_INFO = new BasicArrayTypeInfo<String[], String>(String[].class, BasicTypeInfo.STRING_TYPE_INFO);
public static final BasicArrayTypeInfo<Boolean[], Boolean> BOOLEAN_ARRAY_TYPE_INFO = new BasicArrayTypeInfo<Boolean[], Boolean>(Boolean[].class, BasicTypeInfo.BOOLEAN_TYPE_INFO);
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
index 7bf7298..aa0a445 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
@@ -55,6 +55,8 @@ import org.apache.flink.api.common.typeutils.base.VoidSerializer;
*/
public class BasicTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> {
+ private static final long serialVersionUID = 1L;
+
public static final BasicTypeInfo<String> STRING_TYPE_INFO = new BasicTypeInfo<String>(String.class, StringSerializer.INSTANCE, StringComparator.class);
public static final BasicTypeInfo<Boolean> BOOLEAN_TYPE_INFO = new BasicTypeInfo<Boolean>(Boolean.class, BooleanSerializer.INSTANCE, BooleanComparator.class);
public static final BasicTypeInfo<Byte> BYTE_TYPE_INFO = new BasicTypeInfo<Byte>(Byte.class, ByteSerializer.INSTANCE, ByteComparator.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java
index 367670c..184bdff 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/NothingTypeInfo.java
@@ -16,14 +16,19 @@
* limitations under the License.
*/
-
package org.apache.flink.api.common.typeinfo;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.types.Nothing;
+/**
+ * Placeholder type information for the {@link Nothing} type.
+ */
public class NothingTypeInfo extends TypeInformation<Nothing> {
+
+ private static final long serialVersionUID = 1L;
+
@Override
public boolean isBasicType() {
return false;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
index 56e2680..3ae06a3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
@@ -33,7 +33,15 @@ import org.apache.flink.api.common.typeutils.base.array.IntPrimitiveArraySeriali
import org.apache.flink.api.common.typeutils.base.array.LongPrimitiveArraySerializer;
import org.apache.flink.api.common.typeutils.base.array.ShortPrimitiveArraySerializer;
+/**
+ * A {@link TypeInformation} for arrays of primitive types (int, long, double, ...).
+ * Supports the creation of dedicated efficient serializers for these types.
+ *
+ * @param <T> The type represented by this type information, e.g., int[], double[], long[]
+ */
public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> {
+
+ private static final long serialVersionUID = 1L;
public static final PrimitiveArrayTypeInfo<boolean[]> BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO = new PrimitiveArrayTypeInfo<boolean[]>(boolean[].class, BooleanPrimitiveArraySerializer.INSTANCE);
public static final PrimitiveArrayTypeInfo<byte[]> BYTE_PRIMITIVE_ARRAY_TYPE_INFO = new PrimitiveArrayTypeInfo<byte[]>(byte[].class, BytePrimitiveArraySerializer.INSTANCE);
@@ -46,10 +54,24 @@ public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> {
// --------------------------------------------------------------------------------------------
+ /** The class of the array (such as int[].class) */
private final Class<T> arrayClass;
+
+ /** The serializer for the array */
private final TypeSerializer<T> serializer;
+ /**
+ * Creates a new type info for a
+ * @param arrayClass The class of the array (such as int[].class)
+ * @param serializer The serializer for the array.
+ */
private PrimitiveArrayTypeInfo(Class<T> arrayClass, TypeSerializer<T> serializer) {
+ if (arrayClass == null || serializer == null) {
+ throw new NullPointerException();
+ }
+ if (!(arrayClass.isArray() && arrayClass.getComponentType().isPrimitive())) {
+ throw new IllegalArgumentException("Class must represent an array of primitives.");
+ }
this.arrayClass = arrayClass;
this.serializer = serializer;
}
@@ -98,6 +120,14 @@ public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> {
// --------------------------------------------------------------------------------------------
+ /**
+ * Tries to get the PrimitiveArrayTypeInfo for an array. Returns null, if the type is an array,
+ * but the component type is not a primitive type.
+ *
+ * @param type The class of the array.
+ * @return The corresponding PrimitiveArrayTypeInfo, or null, if the array is not an array of primitives.
+ * @throws InvalidTypesException Thrown, if the given class does not represent an array.
+ */
@SuppressWarnings("unchecked")
public static <X> PrimitiveArrayTypeInfo<X> getInfoFor(Class<X> type) {
if (!type.isArray()) {
@@ -108,8 +138,10 @@ public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> {
return (PrimitiveArrayTypeInfo<X>) TYPES.get(type);
}
+ /** Static map from array class to type info */
private static final Map<Class<?>, PrimitiveArrayTypeInfo<?>> TYPES = new HashMap<Class<?>, PrimitiveArrayTypeInfo<?>>();
+ // initialization of the static map
static {
TYPES.put(boolean[].class, BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO);
TYPES.put(byte[].class, BYTE_PRIMITIVE_ARRAY_TYPE_INFO);
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
index 0ed5afd..beea169 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
@@ -23,23 +23,110 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import java.io.Serializable;
+/**
+ * TypeInformation is the core class of Flink's type system. Flink requires a type information
+ * for all types that are used as input or return type of a user function. This type information
+ * class acts as the tool to generate serializers and comparators, and to perform semantic checks
+ * such as whether the fields that are uses as join/grouping keys actually exist.
+ * <p>
+ * The type information also bridges between the programming languages object model and a
+ * logical flat schema. It maps fields from the types to to columns (fields) in a flat schema.
+ * Not all fields from a type are mapped to a separate fields in the flat schema and
+ * often, entire types are mapped to one field. It is important to notice that the schema must
+ * hold for all instances of a type. For that reason, elements in lists and arrays are not
+ * assigned to individual fields, but the lists and arrays are considered to be one field in total,
+ * to account for different lengths in the arrays.
+ * <ul>
+ * <li>Basic types are indivisible and are considered a single field.</li>
+ * <li>Arrays and collections are one field</li>
+ * <li>Tuples and case classes represent as many fields as the class has fields</li>
+ * <li></li>
+ * <li></li>
+ * <li></li>
+ * </ul>
+ * <p>
+ * To represent this properly, each type has an <i>arity</i> (the number of fields it contains
+ * directly), and a <i>total number of fields</i> (number of fields in the entire schema of this
+ * type, including nested types).
+ * <p>
+ * Consider the example below:
+ * <pre>{@code
+ * public class InnerType {
+ * public int id;
+ * public String text;
+ * }
+ *
+ * public class OuterType {
+ * public long timestamp;
+ * public InnerType nestedType;
+ * }
+ * }</pre>
+ *
+ * The types "id", "text", and "timestamp" are basic types that take up one field. The "InnerType"
+ * has an arity of two, and also two fields totally. The "OuterType" has an arity of two fields,
+ * and a total number of three fields ( it contains "id", "text", and "timestamp" through recursive flattening).
+ *
+ * @param <T> The type represented by this type information.
+ */
public abstract class TypeInformation<T> implements Serializable {
+
+ private static final long serialVersionUID = -7742311969684489493L;
+ /**
+ * Checks if this type information represents a basic type.
+ * Basic types are defined in {@link BasicTypeInfo} and are primitives, their boxing types,
+ * Strings, Date, Void, ...
+ *
+ * @return True, if this type information describes a basic type, false otherwise.
+ */
public abstract boolean isBasicType();
+ /**
+ * Checks if this type information represents a Tuple type.
+ * Tuple types are subclasses of the Java API tuples.
+ *
+ * @return True, if this type information describes a tuple type, false otherwise.
+ */
public abstract boolean isTupleType();
+ /**
+ * Gets the arity of this type - the number of fields without nesting.
+ *
+ * @return Gets the number of fields in this type without nesting.
+ */
public abstract int getArity();
- public abstract Class<T> getTypeClass();
+ /**
+ * Gets the number of logical fields in this type. This includes its nested and transitively nested
+ * fields, in the case of composite types. In the example below, the OuterType type has three
+ * fields in total.
+ *
+ *
+ * @return The number of fields in this type, including its sub-fields (for composite types)
+ */
+ public abstract int getTotalFields();
+ /**
+ * Gets the class of the type represented by this type information.
+ *
+ * @return The class of the type represented by this type information.
+ */
+ public abstract Class<T> getTypeClass();
+
+ /**
+ * Checks whether this type can be used as a key. As a bare minimum, types have
+ * to be hashable and comparable to be keys.
+ *
+ * @return True, if the type can be used as a key, false otherwise.
+ */
public abstract boolean isKeyType();
- public abstract TypeSerializer<T> createSerializer(ExecutionConfig config);
-
/**
- * @return The number of fields in this type, including its sub-fields (for compsite types)
+ * Creates a serializer for the type. The serializer may use the ExecutionConfig
+ * for parameterization.
+ *
+ * @param config The config used to parameterize the serializer.
+ * @return A serializer for this type.
*/
- public abstract int getTotalFields();
-
+ public abstract TypeSerializer<T> createSerializer(ExecutionConfig config);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
index f3e4cfa..17b8dab 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
@@ -33,6 +33,8 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
*/
public abstract class CompositeType<T> extends TypeInformation<T> {
+ private static final long serialVersionUID = 1L;
+
protected final Class<T> typeClass;
public CompositeType(Class<T> typeClass) {
@@ -162,6 +164,8 @@ public abstract class CompositeType<T> extends TypeInformation<T> {
public static class InvalidFieldReferenceException extends IllegalArgumentException {
+ private static final long serialVersionUID = 1L;
+
public InvalidFieldReferenceException(String s) {
super(s);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeComparator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeComparator.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeComparator.java
index 8323dc3..6c9e33e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeComparator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeTypeComparator.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.common.typeutils;
import java.util.LinkedList;
import java.util.List;
+@SuppressWarnings("rawtypes")
public abstract class CompositeTypeComparator<T> extends TypeComparator<T> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
index 0860d41..394aba9 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/StringParser.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.types.parser;
/**
@@ -40,7 +39,6 @@ public class StringParser extends FieldParser<String> {
public int parseField(byte[] bytes, int startPos, int limit, byte[] delimiter, String reusable) {
int i = startPos;
- byte current;
final int delimLimit = limit-delimiter.length+1;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java b/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
index 3cbb21e..086e3e4 100644
--- a/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
+++ b/flink-core/src/main/java/org/apache/flink/types/parser/StringValueParser.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.types.parser;
import org.apache.flink.types.StringValue;
@@ -45,7 +44,6 @@ public class StringValueParser extends FieldParser<StringValue> {
this.result = reusable;
int i = startPos;
- byte current;
final int delimLimit = limit-delimiter.length+1;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java b/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java
index 68e0a09..5518a50 100644
--- a/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java
+++ b/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java
@@ -24,6 +24,8 @@ import java.util.Iterator;
public abstract class SplittableIterator<T> implements Iterator<T>, Serializable {
+ private static final long serialVersionUID = 200377674313072307L;
+
public abstract Iterator<T>[] split(int numPartitions);
public Iterator<T> getSplit(int num, int numPartitions) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
index 8b25249..64cf9da 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred;
import java.io.IOException;
@@ -30,6 +29,8 @@ import org.apache.hadoop.mapred.JobConf;
public class HadoopInputFormat<K, V> extends HadoopInputFormatBase<K, V, Tuple2<K,V>> implements ResultTypeQueryable<Tuple2<K,V>> {
+ private static final long serialVersionUID = 1L;
+
public HadoopInputFormat(org.apache.hadoop.mapred.InputFormat<K,V> mapredInputFormat, Class<K> key, Class<V> value, JobConf job) {
super(mapredInputFormat, key, value, job);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
index 75623e2..44cfb57 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred;
import java.io.IOException;
@@ -26,6 +25,8 @@ import org.apache.hadoop.mapred.JobConf;
public class HadoopOutputFormat<K,V> extends HadoopOutputFormatBase<K, V, Tuple2<K, V>> {
+ private static final long serialVersionUID = 1L;
+
public HadoopOutputFormat(org.apache.hadoop.mapred.OutputFormat<K, V> mapredOutputFormat, JobConf job) {
super(mapredOutputFormat, job);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopInputSplit.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopInputSplit.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopInputSplit.java
index beef5d7..da99404 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopInputSplit.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/wrapper/HadoopInputSplit.java
@@ -16,13 +16,13 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred.wrapper;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.LocatableInputSplit;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -30,11 +30,13 @@ import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.io.WritableFactories;
import org.apache.hadoop.mapred.JobConf;
-
+/**
+ * A wrapper that represents an input split from the Hadoop mapred API as
+ * a Flink {@link InputSplit}.
+ */
public class HadoopInputSplit extends LocatableInputSplit {
private static final long serialVersionUID = 1L;
-
private transient org.apache.hadoop.mapred.InputSplit hadoopInputSplit;
@@ -100,7 +102,6 @@ public class HadoopInputSplit extends LocatableInputSplit {
}
- @SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
this.splitNumber=in.readInt();
this.hadoopInputSplitTypeName = in.readUTF();
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
index efe97f1..4d29d74 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapreduce;
import java.io.IOException;
@@ -30,6 +29,8 @@ import org.apache.flink.api.java.typeutils.TypeExtractor;
public class HadoopInputFormat<K, V> extends HadoopInputFormatBase<K, V, Tuple2<K, V>> implements ResultTypeQueryable<Tuple2<K,V>> {
+ private static final long serialVersionUID = 1L;
+
public HadoopInputFormat(org.apache.hadoop.mapreduce.InputFormat<K,V> mapreduceInputFormat, Class<K> key, Class<V> value, Job job) {
super(mapreduceInputFormat, key, value, job);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
index 7d3675c..26a2c0d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapreduce;
import java.io.IOException;
@@ -26,6 +25,8 @@ import org.apache.hadoop.mapreduce.Job;
public class HadoopOutputFormat<K, V> extends HadoopOutputFormatBase<K, V, Tuple2<K, V>> {
+ private static final long serialVersionUID = 1L;
+
public HadoopOutputFormat(org.apache.hadoop.mapreduce.OutputFormat<K,V> mapreduceOutputFormat, Job job) {
super(mapreduceOutputFormat, job);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
index a7ae428..10e631d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapreduce;
import org.apache.flink.api.common.io.FinalizeOnMaster;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/wrapper/HadoopInputSplit.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/wrapper/HadoopInputSplit.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/wrapper/HadoopInputSplit.java
index f2758b3..16d7f3a 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/wrapper/HadoopInputSplit.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/wrapper/HadoopInputSplit.java
@@ -16,13 +16,13 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapreduce.wrapper;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.LocatableInputSplit;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
@@ -30,9 +30,14 @@ import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableFactories;
import org.apache.hadoop.mapreduce.JobContext;
-
+/**
+ * A wrapper that represents an input split from the Hadoop mapreduce API as
+ * a Flink {@link InputSplit}.
+ */
public class HadoopInputSplit extends LocatableInputSplit {
+ private static final long serialVersionUID = 1L;
+
public transient org.apache.hadoop.mapreduce.InputSplit mapreduceInputSplit;
public transient JobContext jobContext;
@@ -90,7 +95,6 @@ public class HadoopInputSplit extends LocatableInputSplit {
}
- @SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
this.splitNumber=in.readInt();
String className = in.readUTF();
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java
index 291acce..024eb71 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/EnumTypeInfo.java
@@ -26,8 +26,15 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.EnumComparator;
import org.apache.flink.api.common.typeutils.base.EnumSerializer;
+/**
+ * A {@link TypeInformation} for java enumeration types.
+ *
+ * @param <T> The type represented by this type information.
+ */
public class EnumTypeInfo<T extends Enum<T>> extends TypeInformation<T> implements AtomicType<T> {
+ private static final long serialVersionUID = 8936740290137178660L;
+
private final Class<T> typeClass;
public EnumTypeInfo(Class<T> typeClass) {
@@ -80,6 +87,10 @@ public class EnumTypeInfo<T extends Enum<T>> extends TypeInformation<T> implemen
return new EnumSerializer<T>(typeClass);
}
+ // ------------------------------------------------------------------------
+ // Standard utils
+ // ------------------------------------------------------------------------
+
@Override
public String toString() {
return "EnumTypeInfo<" + typeClass.getName() + ">";
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
index cb0ac31..c629ba3 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
@@ -28,6 +28,8 @@ import org.apache.flink.api.java.typeutils.runtime.KryoSerializer;
public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> {
+ private static final long serialVersionUID = -7959114120287706504L;
+
private final Class<T> typeClass;
public GenericTypeInfo(Class<T> typeClass) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/MissingTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/MissingTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/MissingTypeInfo.java
index f0da264..227c68c 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/MissingTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/MissingTypeInfo.java
@@ -29,6 +29,8 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
*/
public class MissingTypeInfo extends TypeInformation<InvalidTypesException> {
+ private static final long serialVersionUID = -4212082837126702723L;
+
private String functionName;
private InvalidTypesException typeException;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java
index b4bc86b..9be8408 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ObjectArrayTypeInfo.java
@@ -31,6 +31,8 @@ import org.apache.flink.api.common.typeutils.base.GenericArraySerializer;
public class ObjectArrayTypeInfo<T, C> extends TypeInformation<T> {
+ private static final long serialVersionUID = 1L;
+
private final Type arrayType;
private final Type componentType;
private final TypeInformation<C> componentInfo;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
index 503b6f2..e35ed38 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
@@ -39,12 +39,21 @@ import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
import com.google.common.base.Joiner;
-
/**
- * TypeInformation for arbitrary (they have to be java-beans-style) java objects (what we call POJO).
+ * TypeInformation for "Java Beans"-style types. Flink refers to them as POJOs,
+ * since the conditions are slightly different from Java Beans.
+ * A type is considered a FLink POJO type, if it fulfills the conditions below.
+ * <ul>
+ * <li>It is a public class, and standalone (not a non-static inner class)</li>
+ * <li>It has a public no-argument constructor.</li>
+ * <li>All fields are either public, or have public getters and setters.</li>
+ * </ul>
*
+ * @param <T> The type represented by this type information.
*/
public class PojoTypeInfo<T> extends CompositeType<T> {
+
+ private static final long serialVersionUID = 1L;
private final static String REGEX_FIELD = "[\\p{L}_\\$][\\p{L}\\p{Digit}_\\$]*";
private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/RecordTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/RecordTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/RecordTypeInfo.java
index 60fa110..9857eb6 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/RecordTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/RecordTypeInfo.java
@@ -24,12 +24,13 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.record.RecordSerializer;
import org.apache.flink.types.Record;
-
/**
- * Type information for the record API.
+ * Type information for the {@link Record} data type.
*/
public class RecordTypeInfo extends TypeInformation<Record> {
+ private static final long serialVersionUID = 1L;
+
@Override
public boolean isBasicType() {
return false;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/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 c1e573d..e88650d 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
@@ -31,11 +31,15 @@ import org.apache.flink.api.java.tuple.*;
import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
-
-
-
+/**
+ * A {@link TypeInformation} for the tuple types of the Java API.
+ *
+ * @param <T> The type of the tuple.
+ */
public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> {
+ private static final long serialVersionUID = 1L;
+
@SuppressWarnings("unchecked")
public TupleTypeInfo(TypeInformation<?>... types) {
this((Class<T>) Tuple.getTupleClass(types.length), types);
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/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 709a729..5c8a57a 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
@@ -30,6 +30,8 @@ import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
+ private static final long serialVersionUID = 1L;
+
private final static String REGEX_FIELD = "(f?)([0-9]+)";
private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
private final static String REGEX_NESTED_FIELDS_WILDCARD = REGEX_NESTED_FIELDS
@@ -40,6 +42,8 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
private static final Pattern PATTERN_NESTED_FIELDS = Pattern.compile(REGEX_NESTED_FIELDS);
private static final Pattern PATTERN_NESTED_FIELDS_WILDCARD = Pattern.compile(REGEX_NESTED_FIELDS_WILDCARD);
+ // --------------------------------------------------------------------------------------------
+
protected final TypeInformation<?>[] types;
protected final Class<T> tupleType;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
index 1486ee5..8288b45 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/ValueTypeInfo.java
@@ -31,9 +31,16 @@ import org.apache.flink.api.java.typeutils.runtime.ValueSerializer;
import org.apache.flink.types.CopyableValue;
import org.apache.flink.types.Value;
-
+/**
+ * Type information for data types that extend the {@link Value} interface. The value
+ * interface allows types to define their custom serialization and deserialization routines.
+ *
+ * @param <T> The type of the class represented by this type information.
+ */
public class ValueTypeInfo<T extends Value> extends TypeInformation<T> implements AtomicType<T> {
+ private static final long serialVersionUID = 1L;
+
private final Class<T> type;
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java
index 89d7818..a91b888 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/WritableTypeInfo.java
@@ -28,8 +28,16 @@ import org.apache.flink.api.java.typeutils.runtime.WritableComparator;
import org.apache.flink.api.java.typeutils.runtime.WritableSerializer;
import org.apache.hadoop.io.Writable;
+/**
+ * Type information for data types that extend Hadoop's {@link Writable} interface. The Writable
+ * interface defines the serialization and deserialization routines for the data type.
+ *
+ * @param <T> The type of the class represented by this type information.
+ */
public class WritableTypeInfo<T extends Writable> extends TypeInformation<T> implements AtomicType<T> {
+ private static final long serialVersionUID = 1L;
+
private final Class<T> typeClass;
public WritableTypeInfo(Class<T> typeClass) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java
index 2758bd6..fa27c87 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/AvroSerializer.java
@@ -79,8 +79,8 @@ public final class AvroSerializer<T> extends TypeSerializer<T> {
}
@Override
- public AvroSerializer duplicate() {
- return new AvroSerializer(type, typeToInstantiate);
+ public AvroSerializer<T> duplicate() {
+ return new AvroSerializer<T>(type, typeToInstantiate);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
index 6033094..f7742ca 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -58,12 +58,12 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
private int numFields;
- private transient Map<Class<?>, TypeSerializer> subclassSerializerCache;
+ private transient Map<Class<?>, TypeSerializer<?>> subclassSerializerCache;
private transient ClassLoader cl;
private Map<Class<?>, Integer> registeredClasses;
- private TypeSerializer[] registeredSerializers;
+ private TypeSerializer<?>[] registeredSerializers;
private final ExecutionConfig executionConfig;
@@ -87,7 +87,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
cl = Thread.currentThread().getContextClassLoader();
- subclassSerializerCache = new HashMap<Class<?>, TypeSerializer>();
+ subclassSerializerCache = new HashMap<Class<?>, TypeSerializer<?>>();
// We only want those classes that are not our own class and are actually sub-classes.
List<Class<?>> cleanedTaggedClasses = new ArrayList<Class<?>>(registeredPojoTypes.size());
@@ -150,10 +150,10 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
cl = Thread.currentThread().getContextClassLoader();
- subclassSerializerCache = new HashMap<Class<?>, TypeSerializer>();
+ subclassSerializerCache = new HashMap<Class<?>, TypeSerializer<?>>();
}
- private TypeSerializer getSubclassSerializer(Class<?> subclass) {
+ private TypeSerializer<?> getSubclassSerializer(Class<?> subclass) {
TypeSerializer<?> result = subclassSerializerCache.get(subclass);
if (result == null) {
@@ -169,6 +169,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
return result;
}
+ @SuppressWarnings("unused")
private boolean hasField(Field f) {
for (Field field: fields) {
if (f.equals(field)) {
@@ -178,7 +179,6 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
return false;
}
- @SuppressWarnings("unchecked")
private void copyBaseFieldOrder(PojoSerializer<?> baseSerializer) {
// do nothing for now, but in the future, adapt subclass serializer to have same
// ordering as base class serializer so that binary comparison on base class fields
@@ -193,7 +193,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
@Override
public PojoSerializer<T> duplicate() {
boolean stateful = false;
- TypeSerializer[] duplicateFieldSerializers = new TypeSerializer[fieldSerializers.length];
+ TypeSerializer<?>[] duplicateFieldSerializers = new TypeSerializer[fieldSerializers.length];
for (int i = 0; i < fieldSerializers.length; i++) {
duplicateFieldSerializers[i] = fieldSerializers[i].duplicate();
@@ -237,7 +237,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
public T copy(T from) {
if (from == null) {
return null;
@@ -277,7 +277,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
public T copy(T from, T reuse) {
if (from == null) {
return null;
@@ -319,7 +319,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
public void serialize(T value, DataOutputView target) throws IOException {
int flags = 0;
// handle null values
@@ -378,7 +378,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
public T deserialize(DataInputView source) throws IOException {
int flags = source.readByte();
if((flags & IS_NULL) != 0) {
@@ -436,7 +436,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
}
@Override
- @SuppressWarnings("unchecked")
+ @SuppressWarnings({"unchecked", "rawtypes"})
public T deserialize(T reuse, DataInputView source) throws IOException {
// handle null values
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/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 6e5925d..231486d 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
@@ -38,7 +38,7 @@ public final class TupleSerializer<T extends Tuple> extends TupleSerializerBase<
@Override
public TupleSerializer<T> duplicate() {
boolean stateful = false;
- TypeSerializer[] duplicateFieldSerializers = new TypeSerializer[fieldSerializers.length];
+ TypeSerializer<?>[] duplicateFieldSerializers = new TypeSerializer<?>[fieldSerializers.length];
for (int i = 0; i < fieldSerializers.length; i++) {
duplicateFieldSerializers[i] = fieldSerializers[i].duplicate();
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
index 777122e..60012ee 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializer.java
@@ -98,8 +98,8 @@ public class WritableSerializer<T extends Writable> extends TypeSerializer<T> {
}
@Override
- public WritableSerializer duplicate() {
- return new WritableSerializer(typeClass);
+ public WritableSerializer<T> duplicate() {
+ return new WritableSerializer<T>(typeClass);
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
index 89aa67e..63f8df4 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatTest.java
@@ -16,19 +16,15 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred;
-
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapred.FileInputFormat;
import org.junit.Test;
@@ -39,11 +35,9 @@ import static org.junit.Assert.fail;
public class HadoopInputFormatTest {
-
public class DummyVoidKeyInputFormat<T> extends FileInputFormat<Void, T> {
- public DummyVoidKeyInputFormat() {
- }
+ public DummyVoidKeyInputFormat() {}
@Override
public org.apache.hadoop.mapred.RecordReader<Void, T> getRecordReader(org.apache.hadoop.mapred.InputSplit inputSplit, JobConf jobConf, Reporter reporter) throws IOException {
@@ -51,21 +45,17 @@ public class HadoopInputFormatTest {
}
}
-
@Test
public void checkTypeInformation() {
try {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- // Set up the Hadoop Input Format
- Job job = Job.getInstance();
- HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<Void, Long>( new DummyVoidKeyInputFormat(), Void.class, Long.class, new JobConf());
+ HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<Void, Long>(
+ new DummyVoidKeyInputFormat<Long>(), Void.class, Long.class, new JobConf());
TypeInformation<Tuple2<Void,Long>> tupleType = hadoopInputFormat.getProducedType();
TypeInformation<Tuple2<Void,Long>> testTupleType = new TupleTypeInfo<Tuple2<Void,Long>>(BasicTypeInfo.VOID_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
if(tupleType.isTupleType()) {
- if(!((TupleTypeInfo)tupleType).equals(testTupleType)) {
+ if(!((TupleTypeInfo<?>)tupleType).equals(testTupleType)) {
fail("Tuple type information was not set correctly!");
}
} else {
@@ -76,7 +66,5 @@ public class HadoopInputFormatTest {
catch (Exception ex) {
fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage());
}
-
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
index 236d149..919a286 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatTest.java
@@ -16,13 +16,10 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapreduce;
-
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat;
@@ -37,15 +34,11 @@ import java.io.IOException;
import static org.junit.Assert.fail;
-
-
public class HadoopInputFormatTest {
-
public class DummyVoidKeyInputFormat<T> extends FileInputFormat<Void, T> {
- public DummyVoidKeyInputFormat() {
- }
+ public DummyVoidKeyInputFormat() {}
@Override
public RecordReader<Void, T> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
@@ -53,21 +46,19 @@ public class HadoopInputFormatTest {
}
}
-
@Test
public void checkTypeInformation() {
try {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
// Set up the Hadoop Input Format
Job job = Job.getInstance();
- HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<Void, Long>( new DummyVoidKeyInputFormat(), Void.class, Long.class, job);
+ HadoopInputFormat<Void, Long> hadoopInputFormat = new HadoopInputFormat<Void, Long>(
+ new DummyVoidKeyInputFormat<Long>(), Void.class, Long.class, job);
TypeInformation<Tuple2<Void,Long>> tupleType = hadoopInputFormat.getProducedType();
TypeInformation<Tuple2<Void,Long>> testTupleType = new TupleTypeInfo<Tuple2<Void,Long>>(BasicTypeInfo.VOID_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
if(tupleType.isTupleType()) {
- if(!((TupleTypeInfo)tupleType).equals(testTupleType)) {
+ if(!((TupleTypeInfo<?>)tupleType).equals(testTupleType)) {
fail("Tuple type information was not set correctly!");
}
} else {
@@ -78,7 +69,5 @@ public class HadoopInputFormatTest {
catch (Exception ex) {
fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage());
}
-
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
index 34df4eb..54291a8 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/client/JobClient.scala
@@ -172,7 +172,8 @@ object JobClient {
"JobManager address has not been specified in the configuration.")
}
- val hostPort = new InetSocketAddress(InetAddress.getByName(jobManagerAddress), jobManagerRPCPort)
+ val hostPort = new InetSocketAddress(InetAddress.getByName(jobManagerAddress),
+ jobManagerRPCPort)
JobManager.getRemoteJobManagerAkkaURL(hostPort)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 7b84928..5a647d1 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -835,7 +835,8 @@ object TaskManager {
"JobManager address has not been specified in the configuration.")
}
- val hostPort = new InetSocketAddress(InetAddress.getByName(jobManagerAddress), jobManagerRPCPort)
+ val hostPort = new InetSocketAddress(InetAddress.getByName(jobManagerAddress),
+ jobManagerRPCPort)
JobManager.getRemoteJobManagerAkkaURL(hostPort)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6ba5fcdf/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java
index 2ae43f9..595c00b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java
@@ -41,7 +41,7 @@ public class JobManagerStartupTest {
try {
portNum = NetUtils.getAvailablePort();
- portOccupier = new ServerSocket(portNum, 10, InetAddress.getByName("127.0.0.1"));
+ portOccupier = new ServerSocket(portNum, 10, InetAddress.getByName("localhost"));
}
catch (Throwable t) {
// could not find free port, or open a connection there
@@ -49,7 +49,7 @@ public class JobManagerStartupTest {
}
try {
- Tuple2<String, Object> connection = new Tuple2<String, Object>("\"127.0.0.1\"", portNum);
+ Tuple2<String, Object> connection = new Tuple2<String, Object>("localhost", portNum);
JobManager.runJobManager(new Configuration(), ExecutionMode.CLUSTER(), new Some<Tuple2<String, Object>>(connection));
fail("this should throw an exception");
}