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 2016/12/20 15:09:25 UTC
[10/15] flink git commit: [FLINK-5294] Test aggregating aligned
window op restore from 1.1
[FLINK-5294] Test aggregating aligned window op restore from 1.1
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/074d80c8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/074d80c8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/074d80c8
Branch: refs/heads/master
Commit: 074d80c834765f2aa659ea27ff647b99f760a06a
Parents: ac134d6
Author: kl0u <kk...@gmail.com>
Authored: Fri Dec 16 16:14:26 2016 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Tue Dec 20 15:42:54 2016 +0100
----------------------------------------------------------------------
.../windowing/WindowOperatorMigrationTest.java | 98 +++++++++++++++++++
...igration-test-aggr-aligned-flink1.1-snapshot | Bin 0 -> 586 bytes
2 files changed, 98 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/074d80c8/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 c02100d..fd38881 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
@@ -28,6 +28,7 @@ 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.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.LegacyWindowOperatorType;
import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
@@ -566,6 +567,103 @@ public class WindowOperatorMigrationTest {
testHarness.close();
}
+ @Test
+ public void testRestoreAlignedProcessingTimeWindowsFromFlink11() throws Exception {
+ /*
+ TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
+
+ AggregatingProcessingTimeWindowOperator<String, Tuple2<String, Integer>> operator =
+ new AggregatingProcessingTimeWindowOperator<>(
+ new ReduceFunction<Tuple2<String, Integer>>() {
+ private static final long serialVersionUID = -8913160567151867987L;
+
+ @Override
+ public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
+ return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
+ }
+ },
+ new TupleKeySelector(),
+ BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
+ inputType.createSerializer(new ExecutionConfig()),
+ 3000,
+ 3000);
+
+ TestTimeServiceProvider testTimeProvider = new TestTimeServiceProvider();
+
+ OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
+ new OneInputStreamOperatorTestHarness<>(operator, new ExecutionConfig(), testTimeProvider);
+
+ testHarness.configureForKeyedStream(new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ testHarness.open();
+
+ testTimeProvider.setCurrentTime(3);
+
+ // timestamp is ignored in processing time
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), Long.MAX_VALUE));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
+
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
+
+ // do a snapshot, close and restore again
+ StreamTaskState snapshot = testHarness.snapshot(0L, 0L);
+ testHarness.snaphotToFile(snapshot, "src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot");
+ testHarness.close();
+
+ */
+
+ final int WINDOW_SIZE = 3;
+
+ TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
+
+ ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
+ new WindowOperatorTest.SumReducer(),
+ inputType.createSerializer(new ExecutionConfig()));
+
+ WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
+ TumblingProcessingTimeWindows.of(Time.of(WINDOW_SIZE, TimeUnit.SECONDS)),
+ new TimeWindow.Serializer(),
+ new TupleKeySelector(),
+ BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
+ stateDesc,
+ new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
+ ProcessingTimeTrigger.create(),
+ 0,
+ LegacyWindowOperatorType.FAST_AGGREGATING);
+
+ OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
+
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
+
+ testHarness.setup();
+ testHarness.initializeStateFromLegacyCheckpoint("src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot");
+ testHarness.open();
+
+ testHarness.setProcessingTime(5000);
+
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 3), 2999));
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 2), 2999));
+
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
+ testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
+
+ testHarness.setProcessingTime(7000);
+
+ expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 5999));
+
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
+
+ testHarness.close();
+ }
+
private static class TupleKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/074d80c8/flink-streaming-java/src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot b/flink-streaming-java/src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot
new file mode 100644
index 0000000..c086a91
Binary files /dev/null and b/flink-streaming-java/src/test/resources/win-op-migration-test-aggr-aligned-flink1.1-snapshot differ