You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2018/02/06 19:02:42 UTC
[18/31] flink git commit: [FLINK-8421] [DataStream,
tests] Add WindowOperator migration test for Kryo-serialized window
keys
[FLINK-8421] [DataStream, tests] Add WindowOperator migration test for Kryo-serialized window keys
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7a9d7e7a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7a9d7e7a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7a9d7e7a
Branch: refs/heads/master
Commit: 7a9d7e7a243c9170635c6a7cd6c414e16d17b8f8
Parents: 72f0eb1
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Wed Jan 24 17:08:13 2018 +0100
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Tue Feb 6 19:58:59 2018 +0100
----------------------------------------------------------------------
.../operators/windowing/NonPojoType.java | 68 ++++++
.../windowing/WindowOperatorMigrationTest.java | 219 +++++++++++++++----
...n-test-kryo-serialized-key-flink1.2-snapshot | Bin 0 -> 3686 bytes
...n-test-kryo-serialized-key-flink1.3-snapshot | Bin 0 -> 11298 bytes
...n-test-kryo-serialized-key-flink1.4-snapshot | Bin 0 -> 9539 bytes
5 files changed, 241 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7a9d7e7a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/NonPojoType.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/NonPojoType.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/NonPojoType.java
new file mode 100644
index 0000000..e70b847
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/NonPojoType.java
@@ -0,0 +1,68 @@
+/*
+ * 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.streaming.runtime.operators.windowing;
+
+/**
+ * A non-Pojo type used in {@link WindowOperatorMigrationTest} to test
+ * migration behaviours for window operators that have a Kryo-serialized key.
+ *
+ * <p>NOTE: modifying this class would cause tests in {@link WindowOperatorMigrationTest} to fail,
+ * since it is serialized in older version test savepoints.
+ */
+public class NonPojoType implements Comparable<NonPojoType> {
+
+ private final String data;
+
+ NonPojoType(String data) {
+ this.data = data;
+ }
+
+ public String getData() {
+ return data;
+ }
+
+ @Override
+ public int compareTo(NonPojoType o) {
+ return data.compareTo(o.data);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+
+ if (obj == null) {
+ return false;
+ }
+
+ if (!(obj instanceof NonPojoType)) {
+ return false;
+ }
+
+ NonPojoType other = (NonPojoType) obj;
+
+ return data.equals(other.data);
+ }
+
+ @Override
+ public int hashCode() {
+ return data.hashCode();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7a9d7e7a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
index 99bea96..ec537c6 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
@@ -23,11 +23,14 @@ import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
@@ -66,6 +69,7 @@ import java.util.Comparator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -114,7 +118,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple3<String, Long, Long>, TimeWindow> operator = new WindowOperator<>(
EventTimeSessionWindows.withGap(Time.seconds(sessionSize)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<String>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new SessionWindowFunction()),
@@ -123,7 +127,7 @@ public class WindowOperatorMigrationTest {
null /* late data output tag */);
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -160,7 +164,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple3<String, Long, Long>, TimeWindow> operator = new WindowOperator<>(
EventTimeSessionWindows.withGap(Time.seconds(sessionSize)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<String>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new SessionWindowFunction()),
@@ -171,7 +175,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -218,7 +222,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple3<String, Long, Long>, TimeWindow> operator = new WindowOperator<>(
EventTimeSessionWindows.withGap(Time.seconds(sessionSize)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<String>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new SessionWindowFunction()),
@@ -227,7 +231,7 @@ public class WindowOperatorMigrationTest {
null /* late data output tag */);
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -258,7 +262,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple3<String, Long, Long>, TimeWindow> operator = new WindowOperator<>(
EventTimeSessionWindows.withGap(Time.seconds(sessionSize)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<String>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new SessionWindowFunction()),
@@ -269,7 +273,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -322,13 +326,13 @@ public class WindowOperatorMigrationTest {
TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
+ new SumReducer<>(),
inputType.createSerializer(new ExecutionConfig()));
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
@@ -339,7 +343,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -358,11 +362,11 @@ public class WindowOperatorMigrationTest {
testHarness.processWatermark(new Watermark(999));
expectedOutput.add(new Watermark(999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.processWatermark(new Watermark(1999));
expectedOutput.add(new Watermark(1999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
// do snapshot and save to file
OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
@@ -380,13 +384,13 @@ public class WindowOperatorMigrationTest {
TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
+ new SumReducer<>(),
inputType.createSerializer(new ExecutionConfig()));
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
@@ -397,7 +401,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -424,7 +428,7 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 2), 5999));
expectedOutput.add(new Watermark(5999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.close();
}
@@ -444,7 +448,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new RichSumReducer<TimeWindow>()),
@@ -455,7 +459,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -474,11 +478,11 @@ public class WindowOperatorMigrationTest {
testHarness.processWatermark(new Watermark(999));
expectedOutput.add(new Watermark(999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.processWatermark(new Watermark(1999));
expectedOutput.add(new Watermark(1999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
// do snapshot and save to file
OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
@@ -501,7 +505,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new RichSumReducer<TimeWindow>()),
@@ -512,7 +516,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -539,7 +543,7 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 2), 5999));
expectedOutput.add(new Watermark(5999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.close();
}
@@ -554,13 +558,13 @@ public class WindowOperatorMigrationTest {
TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
+ new SumReducer<>(),
inputType.createSerializer(new ExecutionConfig()));
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
@@ -571,7 +575,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -587,7 +591,7 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 1), 2999));
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 1), 2999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
// do snapshot and save to file
OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
@@ -606,13 +610,13 @@ public class WindowOperatorMigrationTest {
TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
+ new SumReducer<>(),
inputType.createSerializer(new ExecutionConfig()));
WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
@@ -623,7 +627,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -645,7 +649,7 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 4), 5999));
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key3", 1), 5999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.close();
}
@@ -665,7 +669,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new RichSumReducer<TimeWindow>()),
@@ -676,7 +680,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
testHarness.open();
@@ -692,7 +696,7 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 1), 2999));
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 1), 2999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
// do snapshot and save to file
OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
@@ -715,7 +719,7 @@ public class WindowOperatorMigrationTest {
WindowOperator<String, Tuple2<String, Integer>, Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
new TimeWindow.Serializer(),
- new TupleKeySelector(),
+ new TupleKeySelector<>(),
BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
stateDesc,
new InternalIterableWindowFunction<>(new RichSumReducer<TimeWindow>()),
@@ -726,7 +730,7 @@ public class WindowOperatorMigrationTest {
ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);
testHarness.setup();
@@ -748,28 +752,151 @@ public class WindowOperatorMigrationTest {
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 4), 5999));
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key3", 1), 5999));
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
testHarness.close();
}
- private static class TupleKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
+ /**
+ * Manually run this to write binary snapshot data.
+ */
+ @Ignore
+ @Test
+ public void writeWindowsWithKryoSerializedKeysSnapshot() throws Exception {
+ final int windowSize = 3;
+
+ TypeInformation<Tuple2<NonPojoType, Integer>> inputType = new TypeHint<Tuple2<NonPojoType, Integer>>() {}.getTypeInfo();
+
+ ReducingStateDescriptor<Tuple2<NonPojoType, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
+ new SumReducer<>(),
+ inputType.createSerializer(new ExecutionConfig()));
+
+ TypeSerializer<NonPojoType> keySerializer = TypeInformation.of(NonPojoType.class).createSerializer(new ExecutionConfig());
+ assertTrue(keySerializer instanceof KryoSerializer);
+
+ WindowOperator<NonPojoType, Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>, TimeWindow> operator = new WindowOperator<>(
+ TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
+ new TimeWindow.Serializer(),
+ new TupleKeySelector<>(),
+ keySerializer,
+ stateDesc,
+ new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<>()),
+ EventTimeTrigger.create(),
+ 0,
+ null /* late data output tag */);
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ OneInputStreamOperatorTestHarness<Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), TypeInformation.of(NonPojoType.class));
+
+ testHarness.setup();
+ testHarness.open();
+
+ // add elements out-of-order
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 1), 3999));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 1), 3000));
+
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key1"), 1), 20));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key1"), 1), 0));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key1"), 1), 999));
+
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 1), 1998));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 1), 1999));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 1), 1000));
+
+ testHarness.processWatermark(new Watermark(999));
+ expectedOutput.add(new Watermark(999));
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
+
+ testHarness.processWatermark(new Watermark(1999));
+ expectedOutput.add(new Watermark(1999));
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
+
+ // do snapshot and save to file
+ OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
+ OperatorSnapshotUtil.writeStateHandle(
+ snapshot,
+ "src/test/resources/win-op-migration-test-kryo-serialized-key-flink" + flinkGenerateSavepointVersion + "-snapshot");
+
+ testHarness.close();
+ }
+
+ @Test
+ public void testRestoreKryoSerializedKeysWindows() throws Exception {
+ final int windowSize = 3;
+
+ TypeInformation<Tuple2<NonPojoType, Integer>> inputType = new TypeHint<Tuple2<NonPojoType, Integer>>() {}.getTypeInfo();
+
+ ReducingStateDescriptor<Tuple2<NonPojoType, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
+ new SumReducer<>(),
+ inputType.createSerializer(new ExecutionConfig()));
+
+ TypeSerializer<NonPojoType> keySerializer = TypeInformation.of(NonPojoType.class).createSerializer(new ExecutionConfig());
+ assertTrue(keySerializer instanceof KryoSerializer);
+
+ WindowOperator<NonPojoType, Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>, TimeWindow> operator = new WindowOperator<>(
+ TumblingEventTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
+ new TimeWindow.Serializer(),
+ new TupleKeySelector<>(),
+ keySerializer,
+ stateDesc,
+ new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<>()),
+ EventTimeTrigger.create(),
+ 0,
+ null /* late data output tag */);
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ OneInputStreamOperatorTestHarness<Tuple2<NonPojoType, Integer>, Tuple2<NonPojoType, Integer>> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector<>(), TypeInformation.of(NonPojoType.class));
+
+ testHarness.setup();
+
+ MigrationTestUtil.restoreFromSnapshot(
+ testHarness,
+ OperatorSnapshotUtil.getResourceFilename(
+ "win-op-migration-test-kryo-serialized-key-flink" + testMigrateVersion + "-snapshot"),
+ testMigrateVersion);
+
+ testHarness.open();
+
+ testHarness.processWatermark(new Watermark(2999));
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>(new NonPojoType("key1"), 3), 2999));
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 3), 2999));
+ expectedOutput.add(new Watermark(2999));
+
+ testHarness.processWatermark(new Watermark(3999));
+ expectedOutput.add(new Watermark(3999));
+
+ testHarness.processWatermark(new Watermark(4999));
+ expectedOutput.add(new Watermark(4999));
+
+ testHarness.processWatermark(new Watermark(5999));
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>(new NonPojoType("key2"), 2), 5999));
+ expectedOutput.add(new Watermark(5999));
+
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator<>());
+ testHarness.close();
+ }
+
+ private static class TupleKeySelector<K> implements KeySelector<Tuple2<K, Integer>, K> {
private static final long serialVersionUID = 1L;
@Override
- public String getKey(Tuple2<String, Integer> value) throws Exception {
+ public K getKey(Tuple2<K, Integer> value) throws Exception {
return value.f0;
}
}
@SuppressWarnings("unchecked")
- private static class Tuple2ResultSortComparator implements Comparator<Object> {
+ private static class Tuple2ResultSortComparator<K extends Comparable> implements Comparator<Object> {
@Override
public int compare(Object o1, Object o2) {
if (o1 instanceof Watermark || o2 instanceof Watermark) {
return 0;
} else {
- StreamRecord<Tuple2<String, Integer>> sr0 = (StreamRecord<Tuple2<String, Integer>>) o1;
- StreamRecord<Tuple2<String, Integer>> sr1 = (StreamRecord<Tuple2<String, Integer>>) o2;
+ StreamRecord<Tuple2<K, Integer>> sr0 = (StreamRecord<Tuple2<K, Integer>>) o1;
+ StreamRecord<Tuple2<K, Integer>> sr1 = (StreamRecord<Tuple2<K, Integer>>) o2;
if (sr0.getTimestamp() != sr1.getTimestamp()) {
return (int) (sr0.getTimestamp() - sr1.getTimestamp());
}
@@ -809,11 +936,11 @@ public class WindowOperatorMigrationTest {
}
}
- private static class SumReducer implements ReduceFunction<Tuple2<String, Integer>> {
+ private static class SumReducer<K> implements ReduceFunction<Tuple2<K, Integer>> {
private static final long serialVersionUID = 1L;
@Override
- public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1,
- Tuple2<String, Integer> value2) throws Exception {
+ public Tuple2<K, Integer> reduce(Tuple2<K, Integer> value1,
+ Tuple2<K, Integer> value2) throws Exception {
return new Tuple2<>(value2.f0, value1.f1 + value2.f1);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7a9d7e7a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.2-snapshot
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.2-snapshot b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.2-snapshot
new file mode 100644
index 0000000..3a4cc98
Binary files /dev/null and b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.2-snapshot differ
http://git-wip-us.apache.org/repos/asf/flink/blob/7a9d7e7a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.3-snapshot
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.3-snapshot b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.3-snapshot
new file mode 100644
index 0000000..aa81eb4
Binary files /dev/null and b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.3-snapshot differ
http://git-wip-us.apache.org/repos/asf/flink/blob/7a9d7e7a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.4-snapshot
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.4-snapshot b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.4-snapshot
new file mode 100644
index 0000000..8449113
Binary files /dev/null and b/flink-streaming-java/src/test/resources/win-op-migration-test-kryo-serialized-key-flink1.4-snapshot differ