You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2015/02/10 14:11:13 UTC
[3/5] flink git commit: [FLINK-1369] [types] Add support for
Subclasses, Interfaces, Abstract Classes.
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
index 51d4b0e..0aa6097 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.common.operators.base;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.RichCoGroupFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -66,16 +67,19 @@ public class CoGroupOperatorCollectionTest implements Serializable {
.build()
);
- final RuntimeContext ctx = new RuntimeUDFContext("Test UDF", 4, 0, null);
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ final RuntimeContext ctx = new RuntimeUDFContext("Test UDF", 4, 0, null, executionConfig);
{
SumCoGroup udf1 = new SumCoGroup();
SumCoGroup udf2 = new SumCoGroup();
-
+
+ executionConfig.disableObjectReuse();
List<Tuple2<String, Integer>> resultSafe = getCoGroupOperator(udf1)
- .executeOnCollections(input1, input2, ctx, true);
+ .executeOnCollections(input1, input2, ctx, executionConfig);
+ executionConfig.enableObjectReuse();
List<Tuple2<String, Integer>> resultRegular = getCoGroupOperator(udf2)
- .executeOnCollections(input1, input2, ctx, false);
+ .executeOnCollections(input1, input2, ctx, executionConfig);
Assert.assertTrue(udf1.isClosed);
Assert.assertTrue(udf2.isClosed);
@@ -95,13 +99,15 @@ public class CoGroupOperatorCollectionTest implements Serializable {
}
{
+ executionConfig.disableObjectReuse();
List<Tuple2<String, Integer>> resultSafe = getCoGroupOperator(new SumCoGroup())
.executeOnCollections(Collections.<Tuple2<String, Integer>>emptyList(),
- Collections.<Tuple2<String, Integer>>emptyList(), ctx, true);
-
+ Collections.<Tuple2<String, Integer>>emptyList(), ctx, executionConfig);
+
+ executionConfig.enableObjectReuse();
List<Tuple2<String, Integer>> resultRegular = getCoGroupOperator(new SumCoGroup())
.executeOnCollections(Collections.<Tuple2<String, Integer>>emptyList(),
- Collections.<Tuple2<String, Integer>>emptyList(), ctx, false);
+ Collections.<Tuple2<String, Integer>>emptyList(), ctx, executionConfig);
Assert.assertEquals(0, resultSafe.size());
Assert.assertEquals(0, resultRegular.size());
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
index cfca5aa..447c8c5 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.common.operators.base;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.RichGroupReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -79,9 +80,12 @@ public class GroupReduceOperatorTest implements java.io.Serializable {
Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
Integer>("bar", 4)));
-
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, null, true);
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, null, false);
+
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.disableObjectReuse();
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, null, executionConfig);
+ executionConfig.enableObjectReuse();
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, null, executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
Set<Tuple2<String, Integer>> resultSetRegular = new HashSet<Tuple2<String, Integer>>(resultRegular);
@@ -155,8 +159,11 @@ public class GroupReduceOperatorTest implements java.io.Serializable {
Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
Integer>("bar", 4)));
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null), true);
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null), false);
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.disableObjectReuse();
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ executionConfig.enableObjectReuse();
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
index b4ef54f..1b38281 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
@@ -20,10 +20,10 @@ package org.apache.flink.api.common.operators.base;
import static org.junit.Assert.*;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
-import org.apache.flink.api.common.operators.base.JoinOperatorBase;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
@@ -101,8 +101,11 @@ public class JoinOperatorBaseTest implements Serializable {
));
try {
- List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null), true);
- List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null), false);
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.disableObjectReuse();
+ List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig), executionConfig);
+ executionConfig.enableObjectReuse();
+ List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig), executionConfig);
assertEquals(expected, new HashSet<Tuple2<Double, String>>(resultSafe));
assertEquals(expected, new HashSet<Tuple2<Double, String>>(resultRegular));
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
index 90bbe41..4e1eebd 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.common.operators.base;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -68,8 +69,11 @@ public class ReduceOperatorTest implements java.io.Serializable {
Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
Integer>("bar", 4)));
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, null, true);
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, null, false);
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.disableObjectReuse();
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, null, executionConfig);
+ executionConfig.enableObjectReuse();
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, null, executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
Set<Tuple2<String, Integer>> resultSetRegular = new HashSet<Tuple2<String, Integer>>(resultRegular);
@@ -132,8 +136,11 @@ public class ReduceOperatorTest implements java.io.Serializable {
Integer>("foo", 3), new Tuple2<String, Integer>("bar", 2), new Tuple2<String,
Integer>("bar", 4)));
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null), true);
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null), false);
+ ExecutionConfig executionConfig = new ExecutionConfig();
+ executionConfig.disableObjectReuse();
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ executionConfig.enableObjectReuse();
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
Set<Tuple2<String, Integer>> resultSetRegular = new HashSet<Tuple2<String, Integer>>(resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
index 118e707..cd495b5 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/io/CollectionInputFormatTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -91,7 +92,7 @@ public class CollectionInputFormatTest {
TypeInformation<ElementType> info = (TypeInformation<ElementType>) TypeExtractor.createTypeInfo(ElementType.class);
CollectionInputFormat<ElementType> inputFormat = new CollectionInputFormat<ElementType>(inputCollection,
- info.createSerializer());
+ info.createSerializer(new ExecutionConfig()));
ByteArrayOutputStream buffer = new ByteArrayOutputStream();
ObjectOutputStream out = new ObjectOutputStream(buffer);
@@ -168,7 +169,7 @@ public class CollectionInputFormatTest {
try {
List<String> inputCollection = Arrays.asList(data);
- CollectionInputFormat<String> inputFormat = new CollectionInputFormat<String>(inputCollection, BasicTypeInfo.STRING_TYPE_INFO.createSerializer());
+ CollectionInputFormat<String> inputFormat = new CollectionInputFormat<String>(inputCollection, BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()));
// serialize
ByteArrayOutputStream baos = new ByteArrayOutputStream();
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/io/TypeSerializerFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/io/TypeSerializerFormatTest.java b/flink-java/src/test/java/org/apache/flink/api/java/io/TypeSerializerFormatTest.java
index 7dd1135..f6e3c2a 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/io/TypeSerializerFormatTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/io/TypeSerializerFormatTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.io;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.io.BinaryInputFormat;
import org.apache.flink.api.common.io.BinaryOutputFormat;
import org.apache.flink.api.common.io.BlockInfo;
@@ -51,7 +52,7 @@ public class TypeSerializerFormatTest extends SequentialFormatTestBase<Tuple2<In
resultType = TypeExtractor.getForObject(getRecord(0));
- serializer = resultType.createSerializer();
+ serializer = resultType.createSerializer(new ExecutionConfig());
}
@Before
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/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 b96b3bc..dade55c 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
@@ -1077,10 +1077,16 @@ public class TypeExtractorTest {
public interface Testable {}
- public abstract class AbstractClass {}
+ public static abstract class AbstractClassWithoutMember {}
+
+ public static abstract class AbstractClassWithMember {
+ public int x;
+ }
@Test
- public void testAbstractAndInterfaceTypesException() {
+ public void testAbstractAndInterfaceTypes() {
+
+ // interface
RichMapFunction<String, ?> function = new RichMapFunction<String, Testable>() {
private static final long serialVersionUID = 1L;
@@ -1089,21 +1095,35 @@ public class TypeExtractorTest {
return null;
}
};
-
- TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, BasicTypeInfo.STRING_TYPE_INFO, null, true);
+
+ TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(function, BasicTypeInfo.STRING_TYPE_INFO);
+ Assert.assertTrue(ti instanceof GenericTypeInfo);
+
+ // abstract class with out class member
+ RichMapFunction<String, ?> function2 = new RichMapFunction<String, AbstractClassWithoutMember>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public AbstractClassWithoutMember map(String value) throws Exception {
+ return null;
+ }
+ };
+
+ ti = TypeExtractor.getMapReturnTypes(function2, BasicTypeInfo.STRING_TYPE_INFO);
Assert.assertTrue(ti instanceof GenericTypeInfo);
- RichMapFunction<String, ?> function2 = new RichMapFunction<String, AbstractClass>() {
+ // abstract class with class member
+ RichMapFunction<String, ?> function3 = new RichMapFunction<String, AbstractClassWithMember>() {
private static final long serialVersionUID = 1L;
@Override
- public AbstractClass map(String value) throws Exception {
+ public AbstractClassWithMember map(String value) throws Exception {
return null;
}
};
- TypeInformation<?> ti2 = TypeExtractor.getMapReturnTypes(function2, BasicTypeInfo.STRING_TYPE_INFO, null, true);
- Assert.assertTrue(ti2 instanceof GenericTypeInfo);
+ ti = TypeExtractor.getMapReturnTypes(function3, BasicTypeInfo.STRING_TYPE_INFO);
+ Assert.assertTrue(ti instanceof PojoTypeInfo);
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -1727,7 +1747,7 @@ public class TypeExtractorTest {
+ "myField1=String,myField2=int"
+ ">[][][]"));
Assert.assertEquals("ObjectArrayTypeInfo<ObjectArrayTypeInfo<ObjectArrayTypeInfo<"
- + "PojoType<org.apache.flink.api.java.type.extractor.TypeExtractorTest.CustomType, fields = [myField1: String, myField2: Integer]>"
+ + "PojoType<org.apache.flink.api.java.type.extractor.TypeExtractorTest$CustomType, fields = [myField1: String, myField2: Integer]>"
+ ">>>", ti.toString());
// generic array
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
index c71625a..33aa449 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
@@ -52,7 +52,8 @@ public class CompositeTypeTest {
inNestedTuple3,
BasicTypeInfo.INT_TYPE_INFO );
- private final PojoTypeInfo<?> pojoTypeInfo = ((PojoTypeInfo<?>)TypeExtractor.getForClass(MyPojo.class));
+ private final PojoTypeInfo<?> pojoTypeInfo = ((PojoTypeInfo<?>) TypeExtractor.getForClass
+ (MyPojo.class));
private final TupleTypeInfo<?> pojoInTupleTypeInfo = new TupleTypeInfo<Tuple2<Integer, MyPojo>>(BasicTypeInfo.INT_TYPE_INFO, pojoTypeInfo);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/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 c2f3737..eadf96d 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
@@ -28,12 +28,6 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
-import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.api.java.typeutils.TypeInfoParser;
-import org.apache.flink.api.java.typeutils.ValueTypeInfo;
-import org.apache.flink.api.java.typeutils.WritableTypeInfo;
import org.apache.flink.types.BooleanValue;
import org.apache.flink.types.ByteValue;
import org.apache.flink.types.CharValue;
@@ -300,7 +294,7 @@ public class TypeInfoParserTest {
// pojos
ti = TypeInfoParser.parse("org.apache.flink.api.java.typeutils.TypeInfoParserTest$MyPojo<basic=String>[][][]");
Assert.assertEquals("ObjectArrayTypeInfo<ObjectArrayTypeInfo<ObjectArrayTypeInfo<"
- + "PojoType<org.apache.flink.api.java.typeutils.TypeInfoParserTest.MyPojo, fields = [basic: String]>"
+ + "PojoType<org.apache.flink.api.java.typeutils.TypeInfoParserTest$MyPojo, fields = [basic: String]>"
+ ">>>", ti.toString());
// basic types
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueComparatorTest.java
index a17f499..b317275 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueComparatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/CopyableValueComparatorTest.java
@@ -21,8 +21,6 @@ package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.api.common.typeutils.ComparatorTestBase;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueComparator;
-import org.apache.flink.api.java.typeutils.runtime.CopyableValueSerializer;
import org.apache.flink.types.StringValue;
public class CopyableValueComparatorTest extends ComparatorTestBase<StringValue> {
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
index c190727..d1163d5 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericArraySerializerTest.java
@@ -18,11 +18,12 @@
package org.apache.flink.api.java.typeutils.runtime;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
public class KryoGenericArraySerializerTest extends AbstractGenericArraySerializerTest {
@Override
protected <T> TypeSerializer<T> createComponentSerializer(Class<T> type) {
- return new KryoSerializer<T>(type);
+ return new KryoSerializer<T>(type, new ExecutionConfig());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
index 37dba4e..01c76d9 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeComparatorTest.java
@@ -18,11 +18,12 @@
package org.apache.flink.api.java.typeutils.runtime;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
public class KryoGenericTypeComparatorTest extends AbstractGenericTypeComparatorTest {
@Override
protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
- return new KryoSerializer<T>(type);
+ return new KryoSerializer<T>(type, new ExecutionConfig());
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
index 5953599..8630d95 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoGenericTypeSerializerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.api.java.typeutils.runtime;
import static org.junit.Assert.*;
+import org.apache.flink.api.common.ExecutionConfig;
import org.junit.Test;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -70,7 +71,7 @@ public class KryoGenericTypeSerializerTest extends AbstractGenericTypeSerializer
@Override
protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
- return new KryoSerializer<T>(type);
+ return new KryoSerializer<T>(type, new ExecutionConfig());
}
/**
@@ -94,7 +95,7 @@ public class KryoGenericTypeSerializerTest extends AbstractGenericTypeSerializer
// construct a memory target that is too small for the string
TestDataOutputSerializer target = new TestDataOutputSerializer(10000, 30000);
- KryoSerializer<String> serializer = new KryoSerializer<String>(String.class);
+ KryoSerializer<String> serializer = new KryoSerializer<String>(String.class, new ExecutionConfig());
try {
serializer.serialize(str, target);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoVersusAvroMinibenchmark.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoVersusAvroMinibenchmark.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoVersusAvroMinibenchmark.java
index 4c6b39f..b2ed219 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoVersusAvroMinibenchmark.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoVersusAvroMinibenchmark.java
@@ -27,6 +27,10 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Random;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.MemoryUtils;
@@ -84,9 +88,12 @@ public class KryoVersusAvroMinibenchmark {
System.out.println("Kryo serializer");
{
final TestDataOutputSerializer outView = new TestDataOutputSerializer(100000000);
- final KryoSerializer<MyType> serializer = new KryoSerializer<MyType>(MyType.class);
- serializer.getKryo().register(Tuple2.class);
-
+ ExecutionConfig conf = new ExecutionConfig();
+ conf.registerKryoType(MyType.class);
+ conf.enableForceKryo();
+ TypeInformation<MyType> typeInfo = new GenericTypeInfo<MyType>(MyType.class);
+ final TypeSerializer<MyType> serializer = typeInfo.createSerializer(conf);
+
long start = System.nanoTime();
for (int k = 0; k < NUM_ELEMENTS; k++) {
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoWithCustomSerializersTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoWithCustomSerializersTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoWithCustomSerializersTest.java
index 9d7ab61..7020d80 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoWithCustomSerializersTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/KryoWithCustomSerializersTest.java
@@ -21,7 +21,10 @@ package org.apache.flink.api.java.typeutils.runtime;
import java.util.Collection;
import java.util.HashSet;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.joda.time.LocalDate;
import org.junit.Test;
@@ -40,15 +43,16 @@ public class KryoWithCustomSerializersTest extends AbstractGenericTypeSerializer
b.add(new LocalDate(1L));
b.add(new LocalDate(2L));
-
- KryoSerializer.registerSerializer(LocalDate.class, LocalDateSerializer.class);
-
+
runTests(b);
}
@Override
protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
- return new KryoSerializer<T>(type);
+ ExecutionConfig conf = new ExecutionConfig();
+ conf.registerKryoSerializer(LocalDate.class, LocalDateSerializer.class);
+ TypeInformation<T> typeInfo = new GenericTypeInfo<T>(type);
+ return typeInfo.createSerializer(conf);
}
public static final class LocalDateSerializer extends Serializer<LocalDate> implements java.io.Serializable {
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/MultidimensionalArraySerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/MultidimensionalArraySerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/MultidimensionalArraySerializerTest.java
index 22b6c76..1c97816 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/MultidimensionalArraySerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/MultidimensionalArraySerializerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.typeutils.runtime;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.SerializerTestInstance;
import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -35,7 +36,7 @@ public class MultidimensionalArraySerializerTest {
String[][] array = new String[][]{{null,"b"},{"c","d"},{"e","f"},{"g","h"},null};
TypeInformation<String[][]> ti = TypeExtractor.getForClass(String[][].class);
- SerializerTestInstance<String[][]> testInstance = new SerializerTestInstance<String[][]>(ti.createSerializer(), String[][].class, -1, array);
+ SerializerTestInstance<String[][]> testInstance = new SerializerTestInstance<String[][]>(ti.createSerializer(new ExecutionConfig()), String[][].class, -1, array);
testInstance.testAll();
}
@@ -44,7 +45,7 @@ public class MultidimensionalArraySerializerTest {
int[][] array = new int[][]{{12,1},{48,42},{23,80},{484,849},{987,4}};
TypeInformation<int[][]> ti = TypeExtractor.getForClass(int[][].class);
- SerializerTestInstance<int[][]> testInstance = new SerializerTestInstance<int[][]>(ti.createSerializer(), int[][].class, -1, array);
+ SerializerTestInstance<int[][]> testInstance = new SerializerTestInstance<int[][]>(ti.createSerializer(new ExecutionConfig()), int[][].class, -1, array);
testInstance.testAll();
}
@@ -73,13 +74,13 @@ public class MultidimensionalArraySerializerTest {
Integer[][] array = new Integer[][]{{0,1}, null, {null, 42}};
TypeInformation<Integer[][]> ti = TypeExtractor.getForClass(Integer[][].class);
- SerializerTestInstance<Integer[][]> testInstance = new SerializerTestInstance<Integer[][]>(ti.createSerializer(), Integer[][].class, -1, array);
+ SerializerTestInstance<Integer[][]> testInstance = new SerializerTestInstance<Integer[][]>(ti.createSerializer(new ExecutionConfig()), Integer[][].class, -1, array);
testInstance.testAll();
MyPojo[][] array2 = new MyPojo[][]{{new MyPojo(null, 42), new MyPojo("test2", -1)}, {null, null}, null};
TypeInformation<MyPojo[][]> ti2 = TypeExtractor.getForClass(MyPojo[][].class);
- SerializerTestInstance<MyPojo[][]> testInstance2 = new SerializerTestInstance<MyPojo[][]>(ti2.createSerializer(), MyPojo[][].class, -1, array2);
+ SerializerTestInstance<MyPojo[][]> testInstance2 = new SerializerTestInstance<MyPojo[][]>(ti2.createSerializer(new ExecutionConfig()), MyPojo[][].class, -1, array2);
testInstance2.testAll();
}
@@ -112,7 +113,7 @@ public class MultidimensionalArraySerializerTest {
};
TypeInformation ti = TypeInfoParser.parse("org.apache.flink.api.java.typeutils.runtime.MultidimensionalArraySerializerTest$MyGenericPojo<field=String[][]>[][]");
- SerializerTestInstance testInstance = new SerializerTestInstance(ti.createSerializer(), MyGenericPojo[][].class, -1, (Object) array);
+ SerializerTestInstance testInstance = new SerializerTestInstance(ti.createSerializer(new ExecutionConfig()), MyGenericPojo[][].class, -1, (Object) array);
testInstance.testAll();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java
index e53f48a..1baf443 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoComparatorTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.api.java.typeutils.runtime;
import java.util.Arrays;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.ComparatorTestBase;
import org.apache.flink.api.common.typeutils.CompositeType;
@@ -47,12 +48,12 @@ public class PojoComparatorTest extends ComparatorTestBase<PojoContainingTuple>
ExpressionKeys<PojoContainingTuple> keys = new ExpressionKeys<PojoContainingTuple>(new String[] {"theTuple.*"}, cType);
boolean[] orders = new boolean[keys.getNumberOfKeyFields()];
Arrays.fill(orders, ascending);
- return cType.createComparator(keys.computeLogicalKeyPositions(), orders, 0);
+ return cType.createComparator(keys.computeLogicalKeyPositions(), orders, 0, new ExecutionConfig());
}
@Override
protected TypeSerializer<PojoContainingTuple> createSerializer() {
- return type.createSerializer();
+ return type.createSerializer(new ExecutionConfig());
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoGenericTypeSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoGenericTypeSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoGenericTypeSerializerTest.java
index a176178..d405412 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoGenericTypeSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoGenericTypeSerializerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.api.java.typeutils.runtime;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -27,6 +28,6 @@ public class PojoGenericTypeSerializerTest extends AbstractGenericTypeSerializer
@Override
protected <T> TypeSerializer<T> createSerializer(Class<T> type) {
TypeInformation<T> typeInfo = TypeExtractor.getForClass(type);
- return typeInfo.createSerializer();
+ return typeInfo.createSerializer(new ExecutionConfig());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
index 006625e..1fa7163 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializerTest.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Random;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeComparator;
@@ -40,14 +41,14 @@ import org.junit.Test;
import com.google.common.base.Objects;
/**
- * A test for the {@link org.apache.flink.api.java.typeutils.runtime.PojoSerializer}.
+ * A test for the {@link PojoSerializer}.
*/
public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.TestUserClass> {
private TypeInformation<TestUserClass> type = TypeExtractor.getForClass(TestUserClass.class);
@Override
protected TypeSerializer<TestUserClass> createSerializer() {
- TypeSerializer<TestUserClass> serializer = type.createSerializer();
+ TypeSerializer<TestUserClass> serializer = type.createSerializer(new ExecutionConfig());
assert(serializer instanceof PojoSerializer);
return serializer;
}
@@ -199,14 +200,14 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
pType.getFlatFields("nestedClass.dumm2", 0, result);
int[] fields = new int[1]; // see below
fields[0] = result.get(0).getPosition();
- TypeComparator<TestUserClass> pojoComp = pType.createComparator( fields, new boolean[]{true}, 0);
+ TypeComparator<TestUserClass> pojoComp = pType.createComparator( fields, new boolean[]{true}, 0, new ExecutionConfig());
TestUserClass pojoTestRecord = new TestUserClass(0, "abc", 3d, new int[] {1,2,3}, new NestedTestUserClass(1, "haha", 4d, new int[] {5,4,3}));
int pHash = pojoComp.hash(pojoTestRecord);
Tuple1<String> tupleTest = new Tuple1<String>("haha");
TupleTypeInfo<Tuple1<String>> tType = (TupleTypeInfo<Tuple1<String>>)TypeExtractor.getForObject(tupleTest);
- TypeComparator<Tuple1<String>> tupleComp = tType.createComparator(new int[] {0}, new boolean[] {true}, 0);
+ TypeComparator<Tuple1<String>> tupleComp = tType.createComparator(new int[] {0}, new boolean[] {true}, 0, new ExecutionConfig());
int tHash = tupleComp.hash(tupleTest);
@@ -223,12 +224,12 @@ public class PojoSerializerTest extends SerializerTestBase<PojoSerializerTest.Te
e.printStackTrace();
Assert.fail("Keys must be compatible: "+e.getMessage());
}
- TypeComparator<TestUserClass> multiPojoComp = pType.createComparator( expressKey.computeLogicalKeyPositions(), new boolean[]{true, true, true}, 0);
+ TypeComparator<TestUserClass> multiPojoComp = pType.createComparator( expressKey.computeLogicalKeyPositions(), new boolean[]{true, true, true}, 0, new ExecutionConfig());
int multiPojoHash = multiPojoComp.hash(pojoTestRecord);
// pojo order is: dumm2 (str), dumm1 (int), dumm3 (double).
- TypeComparator<Tuple3<Integer, String, Double>> multiTupleComp = multiTupleType.createComparator(fieldKey.computeLogicalKeyPositions(), new boolean[] {true, true,true}, 0);
+ TypeComparator<Tuple3<Integer, String, Double>> multiTupleComp = multiTupleType.createComparator(fieldKey.computeLogicalKeyPositions(), new boolean[] {true, true,true}, 0, new ExecutionConfig());
int multiTupleHash = multiTupleComp.hash(multiTupleTest);
Assert.assertTrue("The hashing for tuples and pojos must be the same, so that they are mixable. Also for those with multiple key fields", multiPojoHash == multiTupleHash);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java
new file mode 100644
index 0000000..3a03683
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassComparatorTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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 org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.ComparatorTestBase;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.operators.Keys.ExpressionKeys;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.junit.Assert;
+
+import java.util.Arrays;
+
+
+public class PojoSubclassComparatorTest extends ComparatorTestBase<PojoContainingTuple> {
+ TypeInformation<PojoContainingTuple> type = TypeExtractor.getForClass(PojoContainingTuple.class);
+
+ PojoContainingTuple[] data = new PojoContainingTuple[]{
+ new Subclass(1, 1L, 1L, 17L),
+ new Subclass(2, 2L, 2L, 42L),
+ new Subclass(8519, 85190L, 85190L, 117L),
+ new Subclass(8520, 85191L, 85191L, 93L),
+ };
+
+ @Override
+ protected TypeComparator<PojoContainingTuple> createComparator(boolean ascending) {
+ Assert.assertTrue(type instanceof CompositeType);
+ CompositeType<PojoContainingTuple> cType = (CompositeType<PojoContainingTuple>) type;
+ ExpressionKeys<PojoContainingTuple> keys = new ExpressionKeys<PojoContainingTuple>(new String[] {"theTuple.*"}, cType);
+ boolean[] orders = new boolean[keys.getNumberOfKeyFields()];
+ Arrays.fill(orders, ascending);
+ return cType.createComparator(keys.computeLogicalKeyPositions(), orders, 0, new ExecutionConfig());
+ }
+
+ @Override
+ protected TypeSerializer<PojoContainingTuple> createSerializer() {
+ return type.createSerializer(new ExecutionConfig());
+ }
+
+ @Override
+ protected PojoContainingTuple[] getSortedTestData() {
+ return data;
+ }
+
+ public static class Subclass extends PojoContainingTuple {
+
+ public long additionalField;
+
+ public Subclass() {
+ }
+
+ public Subclass(int i, long l1, long l2, long additionalField) {
+ super(i, l1, l2);
+ this.additionalField = additionalField;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassSerializerTest.java
new file mode 100644
index 0000000..8c61a19
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/PojoSubclassSerializerTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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 com.google.common.base.Objects;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.junit.Test;
+
+import java.util.Random;
+
+/**
+ * A test for the {@link PojoSerializer}.
+ */
+public class PojoSubclassSerializerTest extends SerializerTestBase<PojoSubclassSerializerTest.TestUserClassBase> {
+ private TypeInformation<TestUserClassBase> type = TypeExtractor.getForClass(TestUserClassBase.class);
+
+ @Override
+ protected TypeSerializer<TestUserClassBase> createSerializer() {
+ // only register one of the three child classes, the third child class is NO POJO
+ ExecutionConfig conf = new ExecutionConfig();
+ conf.registerPojoType(TestUserClass1.class);
+ TypeSerializer<TestUserClassBase> serializer = type.createSerializer(conf);
+ assert(serializer instanceof PojoSerializer);
+ return serializer;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<TestUserClassBase> getTypeClass() {
+ return TestUserClassBase.class;
+ }
+
+ @Override
+ protected TestUserClassBase[] getTestData() {
+ Random rnd = new Random(874597969123412341L);
+
+ return new TestUserClassBase[]{
+ new TestUserClass1(rnd.nextInt(), "foo", rnd.nextLong()),
+ new TestUserClass2(rnd.nextInt(), "bar", rnd.nextFloat()),
+ new TestUserClass3(rnd.nextInt(), "bar", rnd.nextFloat())
+ };
+
+ }
+
+ @Override
+ @Test
+ public void testInstantiate() {
+ // don't do anything, since the PojoSerializer with subclass will return null
+ }
+
+ // User code class for testing the serializer
+ public static abstract class TestUserClassBase {
+ public int dumm1;
+ public String dumm2;
+
+
+ public TestUserClassBase() {
+ }
+
+ public TestUserClassBase(int dumm1, String dumm2) {
+ this.dumm1 = dumm1;
+ this.dumm2 = dumm2;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(dumm1, dumm2);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClassBase)) {
+ return false;
+ }
+ TestUserClassBase otherTUC = (TestUserClassBase) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ public static class TestUserClass1 extends TestUserClassBase {
+ public long dumm3;
+
+ public TestUserClass1() {
+ }
+
+ public TestUserClass1(int dumm1, String dumm2, long dumm3) {
+ super(dumm1, dumm2);
+ this.dumm3 = dumm3;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClass1)) {
+ return false;
+ }
+ TestUserClass1 otherTUC = (TestUserClass1) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ if (dumm3 != otherTUC.dumm3) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ public static class TestUserClass2 extends TestUserClassBase {
+ public float dumm4;
+
+ public TestUserClass2() {
+ }
+
+ public TestUserClass2(int dumm1, String dumm2, float dumm4) {
+ super(dumm1, dumm2);
+ this.dumm4 = dumm4;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClass2)) {
+ return false;
+ }
+ TestUserClass2 otherTUC = (TestUserClass2) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ if (dumm4 != otherTUC.dumm4) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ public static class TestUserClass3 extends TestUserClassBase {
+ public float dumm4;
+
+ public TestUserClass3(int dumm1, String dumm2, float dumm4) {
+ super(dumm1, dumm2);
+ this.dumm4 = dumm4;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClass3)) {
+ return false;
+ }
+ TestUserClass3 otherTUC = (TestUserClass3) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ if (dumm4 != otherTUC.dumm4) {
+ return false;
+ }
+ return true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/SubclassFromInterfaceSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/SubclassFromInterfaceSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/SubclassFromInterfaceSerializerTest.java
new file mode 100644
index 0000000..efb1b9b
--- /dev/null
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/SubclassFromInterfaceSerializerTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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 com.google.common.base.Objects;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SerializerTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.junit.Test;
+
+import java.util.Random;
+
+/**
+ * Testing the serialization of classes which are subclasses of a class that implements an interface.
+ */
+public class SubclassFromInterfaceSerializerTest extends SerializerTestBase<SubclassFromInterfaceSerializerTest.TestUserInterface> {
+ private TypeInformation<TestUserInterface> type = TypeExtractor.getForClass(TestUserInterface.class);
+
+ @Override
+ protected TypeSerializer<TestUserInterface> createSerializer() {
+ // only register one of the two child classes
+ ExecutionConfig conf = new ExecutionConfig();
+ conf.registerPojoType(TestUserClass2.class);
+ TypeSerializer<TestUserInterface> serializer = type.createSerializer(conf);
+ assert(serializer instanceof KryoSerializer);
+ return serializer;
+ }
+
+ @Override
+ protected int getLength() {
+ return -1;
+ }
+
+ @Override
+ protected Class<TestUserInterface> getTypeClass() {
+ return TestUserInterface.class;
+ }
+
+ @Override
+ protected TestUserInterface[] getTestData() {
+ Random rnd = new Random(874597969123412341L);
+
+ return new TestUserInterface[]{
+ new TestUserClass1(rnd.nextInt(), "foo", rnd.nextLong()),
+ new TestUserClass2(rnd.nextInt(), "bar", rnd.nextFloat())
+ };
+
+ }
+
+ @Override
+ @Test
+ public void testInstantiate() {
+ // don't do anything, since the PojoSerializer with subclass will return null
+ }
+
+ public interface TestUserInterface {}
+
+ // User code class for testing the serializer
+ public static class TestUserClassBase implements TestUserInterface {
+ public int dumm1;
+ public String dumm2;
+
+
+ public TestUserClassBase() {
+ }
+
+ public TestUserClassBase(int dumm1, String dumm2) {
+ this.dumm1 = dumm1;
+ this.dumm2 = dumm2;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(dumm1, dumm2);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClassBase)) {
+ return false;
+ }
+ TestUserClassBase otherTUC = (TestUserClassBase) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ public static class TestUserClass1 extends TestUserClassBase {
+ public long dumm3;
+
+ public TestUserClass1() {
+ }
+
+ public TestUserClass1(int dumm1, String dumm2, long dumm3) {
+ super(dumm1, dumm2);
+ this.dumm3 = dumm3;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClass1)) {
+ return false;
+ }
+ TestUserClass1 otherTUC = (TestUserClass1) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ if (dumm3 != otherTUC.dumm3) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ public static class TestUserClass2 extends TestUserClassBase {
+ public float dumm4;
+
+ public TestUserClass2() {
+ }
+
+ public TestUserClass2(int dumm1, String dumm2, float dumm4) {
+ super(dumm1, dumm2);
+ this.dumm4 = dumm4;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof TestUserClass2)) {
+ return false;
+ }
+ TestUserClass2 otherTUC = (TestUserClass2) other;
+ if (dumm1 != otherTUC.dumm1) {
+ return false;
+ }
+ if (!dumm2.equals(otherTUC.dumm2)) {
+ return false;
+ }
+ if (dumm4 != otherTUC.dumm4) {
+ return false;
+ }
+ return true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD2Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD2Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD2Test.java
index 9f1c7b8..cfc4914 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD2Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD2Test.java
@@ -26,8 +26,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongComparator;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD3Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD3Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD3Test.java
index 8d13ab7..e5a0e6c 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD3Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILD3Test.java
@@ -27,8 +27,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongComparator;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDC3Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDC3Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDC3Test.java
index 82b8b4e..a1e6c40 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDC3Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDC3Test.java
@@ -27,8 +27,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongComparator;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDX1Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDX1Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDX1Test.java
index 25de450..b5c0c1f 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDX1Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDX1Test.java
@@ -25,8 +25,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongComparator;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDXC2Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDXC2Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDXC2Test.java
index 2a5fc29..793a2f4 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDXC2Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorILDXC2Test.java
@@ -26,8 +26,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.LongComparator;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD1Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD1Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD1Test.java
index b2218ad..8cdee9b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD1Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD1Test.java
@@ -25,8 +25,6 @@ import org.apache.flink.api.common.typeutils.base.IntComparator;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD2Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD2Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD2Test.java
index fe152a3..06c292f 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD2Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD2Test.java
@@ -26,8 +26,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringComparator;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD3Test.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD3Test.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD3Test.java
index b92d825..d823a29 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD3Test.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/TupleComparatorISD3Test.java
@@ -27,8 +27,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringComparator;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.typeutils.runtime.TupleComparator;
-import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
import org.apache.flink.api.java.typeutils.runtime.tuple.base.TupleComparatorTestBase;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/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 a701a82..96f8306 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
@@ -22,6 +22,7 @@ package org.apache.flink.api.java.typeutils.runtime;
import java.util.ArrayList;
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.Tuple1;
@@ -209,7 +210,7 @@ public class TupleSerializerTest {
private <T extends Tuple> void runTests(T... instances) {
try {
TupleTypeInfo<T> tupleTypeInfo = (TupleTypeInfo<T>) TypeExtractor.getForObject(instances[0]);
- TypeSerializer<T> serializer = tupleTypeInfo.createSerializer();
+ TypeSerializer<T> serializer = tupleTypeInfo.createSerializer(new ExecutionConfig());
Class<T> tupleClass = tupleTypeInfo.getTypeClass();
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/ValueComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/ValueComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/ValueComparatorTest.java
index e710187..cf9874d 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/ValueComparatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/ValueComparatorTest.java
@@ -21,8 +21,6 @@ package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.api.common.typeutils.ComparatorTestBase;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.typeutils.runtime.ValueComparator;
-import org.apache.flink.api.java.typeutils.runtime.ValueSerializer;
import org.apache.flink.types.StringValue;
public class ValueComparatorTest extends ComparatorTestBase<StringValue> {
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableComparatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableComparatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableComparatorTest.java
index ce1147a..f5a90b7 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableComparatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableComparatorTest.java
@@ -21,8 +21,6 @@ package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.api.common.typeutils.ComparatorTestBase;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.typeutils.runtime.WritableComparator;
-import org.apache.flink.api.java.typeutils.runtime.WritableSerializer;
public class WritableComparatorTest extends ComparatorTestBase<StringArrayWritable> {
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializerTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializerTest.java
index 01349bf..557c0e4 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializerTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/runtime/WritableSerializerTest.java
@@ -18,10 +18,10 @@
package org.apache.flink.api.java.typeutils.runtime;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.SerializerTestInstance;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.WritableTypeInfo;
-import org.apache.flink.api.java.typeutils.runtime.WritableSerializer;
import org.junit.Test;
public class WritableSerializerTest {
@@ -41,7 +41,7 @@ public class WritableSerializerTest {
};
WritableTypeInfo<StringArrayWritable> writableTypeInfo = (WritableTypeInfo<StringArrayWritable>) TypeExtractor.getForObject(data[0]);
- WritableSerializer<StringArrayWritable> writableSerializer = (WritableSerializer<StringArrayWritable>) writableTypeInfo.createSerializer();
+ WritableSerializer<StringArrayWritable> writableSerializer = (WritableSerializer<StringArrayWritable>) writableTypeInfo.createSerializer(new ExecutionConfig());
SerializerTestInstance<StringArrayWritable> testInstance = new SerializerTestInstance<StringArrayWritable>(writableSerializer,writableTypeInfo.getTypeClass(), -1, data);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
----------------------------------------------------------------------
diff --git a/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java b/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
index 49a3fe5..6edd2f7 100644
--- a/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
+++ b/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
@@ -29,7 +29,6 @@ import org.apache.flink.api.common.functions.CrossFunction;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
-import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.MapPartitionFunction;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
index 4f10683..d213069 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.iterative.task;
+import org.apache.flink.api.common.ExecutionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.aggregators.Aggregator;
@@ -165,7 +166,7 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
public DistributedRuntimeUDFContext createRuntimeContext(String taskName) {
Environment env = getEnvironment();
return new IterativeRuntimeUdfContext(taskName, env.getNumberOfSubtasks(),
- env.getIndexInSubtaskGroup(), getUserCodeClassLoader());
+ env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), getExecutionConfig());
}
// --------------------------------------------------------------------------------------------
@@ -355,8 +356,8 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
private class IterativeRuntimeUdfContext extends DistributedRuntimeUDFContext implements IterationRuntimeContext {
- public IterativeRuntimeUdfContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader);
+ public IterativeRuntimeUdfContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
index ff1b272..42bbe3c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
@@ -18,14 +18,21 @@
package org.apache.flink.runtime.jobgraph.tasks;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.util.InstantiationUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Abstract base class for every task class in Flink.
*/
public abstract class AbstractInvokable {
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractInvokable.class);
+
+
/**
* The environment assigned to this invokable.
*/
@@ -111,6 +118,26 @@ public abstract class AbstractInvokable {
}
/**
+ * Returns the global ExecutionConfig.
+ */
+ public ExecutionConfig getExecutionConfig() {
+ try {
+ ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(
+ getJobConfiguration(),
+ ExecutionConfig.CONFIG_KEY,
+ this.getClass().getClassLoader());
+ if (c != null) {
+ return c;
+ } else {
+ return new ExecutionConfig();
+ }
+ } catch (Exception e) {
+ LOG.warn("Could not load ExecutionConfig from Environment, returning default ExecutionConfig: {}", e);
+ return new ExecutionConfig();
+ }
+ }
+
+ /**
* This method is called when a task is canceled either as a result of a user abort or an execution failure. It can
* be overwritten to respond to shut down the user code properly.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
index 330f438..01ab533 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
@@ -74,24 +74,6 @@ public class DataSourceTask<OT> extends AbstractInvokable {
// cancel flag
private volatile boolean taskCanceled = false;
- private ExecutionConfig getExecutionConfig() {
- ExecutionConfig executionConfig = new ExecutionConfig();
- try {
- ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(
- getJobConfiguration(),
- ExecutionConfig.CONFIG_KEY,
- this.getClass().getClassLoader());
- if (c != null) {
- executionConfig = c;
- }
- } catch (IOException e) {
- throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
- }
- return executionConfig;
- }
-
@Override
public void registerInputOutput() {
initInputFormat();
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
index f6d2f4a..ffe09cb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
@@ -45,8 +45,6 @@ public interface PactTaskContext<S, OT> {
TaskConfig getTaskConfig();
- ExecutionConfig getExecutionConfig();
-
ClassLoader getUserCodeClassLoader();
MemoryManager getMemoryManager();
@@ -60,7 +58,9 @@ public interface PactTaskContext<S, OT> {
<X> TypeComparator<X> getDriverComparator(int index);
S getStub();
-
+
+ ExecutionConfig getExecutionConfig();
+
Collector<OT> getOutputCollector();
AbstractInvokable getOwningNepheleTask();
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
index 6b6918d..9118400 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
@@ -1068,7 +1068,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
public DistributedRuntimeUDFContext createRuntimeContext(String taskName) {
Environment env = getEnvironment();
return new DistributedRuntimeUDFContext(taskName, env.getNumberOfSubtasks(),
- env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), env.getCopyTask());
+ env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), getExecutionConfig(), env.getCopyTask());
}
// --------------------------------------------------------------------------------------------
@@ -1081,23 +1081,6 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
}
@Override
- public ExecutionConfig getExecutionConfig() {
- try {
- ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig(
- getOwningNepheleTask().getJobConfiguration(),
- ExecutionConfig.CONFIG_KEY,
- this.getClass().getClassLoader());
- if (c != null) {
- return c;
- } else {
- return new ExecutionConfig();
- }
- } catch (Exception e) {
- throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: " + e);
- }
- }
-
- @Override
public MemoryManager getMemoryManager() {
return getEnvironment().getMemoryManager();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
index 838286e..c84d888 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
@@ -62,7 +62,7 @@ public abstract class ChainedDriver<IT, OT> implements Collector<IT> {
} else {
Environment env = parent.getEnvironment();
this.udfContext = new DistributedRuntimeUDFContext(taskName, env.getNumberOfSubtasks(),
- env.getIndexInSubtaskGroup(), userCodeClassLoader, env.getCopyTask());
+ env.getIndexInSubtaskGroup(), userCodeClassLoader, parent.getExecutionConfig(), env.getCopyTask());
}
this.executionConfig = executionConfig;
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/LargeRecordHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/LargeRecordHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/LargeRecordHandler.java
index 0080d63..ce63021 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/LargeRecordHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/LargeRecordHandler.java
@@ -25,6 +25,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -102,6 +103,8 @@ public class LargeRecordHandler<T> {
private volatile boolean closed;
+ private final ExecutionConfig executionConfig;
+
// --------------------------------------------------------------------------------------------
public LargeRecordHandler(TypeSerializer<T> serializer, TypeComparator<T> comparator,
@@ -115,7 +118,9 @@ public class LargeRecordHandler<T> {
this.memory = checkNotNull(memory);
this.memoryOwner = checkNotNull(memoryOwner);
this.maxFilehandles = maxFilehandles;
-
+
+ this.executionConfig = memoryOwner.getExecutionConfig();
+
checkArgument(maxFilehandles >= 2);
}
@@ -374,13 +379,13 @@ public class LargeRecordHandler<T> {
// --------------------------------------------------------------------------------------------
- private static TypeSerializer<Object> createSerializer(Object key, int pos) {
+ private TypeSerializer<Object> createSerializer(Object key, int pos) {
if (key == null) {
throw new NullKeyFieldException(pos);
}
try {
TypeInformation<Object> info = TypeExtractor.getForObject(key);
- return info.createSerializer();
+ return info.createSerializer(executionConfig);
}
catch (Throwable t) {
throw new RuntimeException("Could not create key serializer for type " + key);
http://git-wip-us.apache.org/repos/asf/flink/blob/7407076d/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
index aa7aec6..e841066 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.FutureTask;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.AbstractRuntimeUDFContext;
@@ -40,12 +41,12 @@ public class DistributedRuntimeUDFContext extends AbstractRuntimeUDFContext {
private final HashMap<String, BroadcastVariableMaterialization<?, ?>> broadcastVars = new HashMap<String, BroadcastVariableMaterialization<?, ?>>();
- public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader);
+ public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
}
- public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, Map<String, FutureTask<Path>> cpTasks) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, cpTasks);
+ public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, FutureTask<Path>> cpTasks) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, cpTasks);
}