/*
 * 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;

import org.apache.flink.api.common.ExecutionConfig;
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.runtime.kryo.KryoSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
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;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableWindowFunction;
import org.apache.flink.streaming.runtime.operators.windowing.functions.InternalSingleValueWindowFunction;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OperatorSnapshotUtil;
import org.apache.flink.streaming.util.TestHarnessUtil;
import org.apache.flink.testutils.migration.MigrationVersion;
import org.apache.flink.util.Collector;

import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.util.Arrays;
import java.util.Collection;
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;

/**
 * Tests for checking whether {@link WindowOperator} can restore from snapshots that were done using
 * previous Flink versions' {@link WindowOperator}.
 *
 * <p>This also checks whether {@link WindowOperator} can restore from a checkpoint of the aligned
 * processing-time windows operator of previous Flink versions.
 *
 * <p>For regenerating the binary snapshot file you have to run the {@code write*()} method on the
 * corresponding Flink release-* branch.
 */
@RunWith(Parameterized.class)
public class WindowOperatorMigrationTest {

    @Parameterized.Parameters(name = "Migration Savepoint: {0}")
    public static Collection<MigrationVersion> parameters() {
        return Arrays.asList(
                MigrationVersion.v1_3,
                MigrationVersion.v1_4,
                MigrationVersion.v1_5,
                MigrationVersion.v1_6,
                MigrationVersion.v1_7,
                MigrationVersion.v1_8,
                MigrationVersion.v1_9,
                MigrationVersion.v1_10,
                MigrationVersion.v1_11,
                MigrationVersion.v1_12,
                MigrationVersion.v1_13);
    }

    private static final TypeInformation<Tuple2<String, Integer>> STRING_INT_TUPLE =
            TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {});

    /**
     * TODO change this to the corresponding savepoint version to be written (e.g. {@link
     * MigrationVersion#v1_3} for 1.3) TODO and remove all @Ignore annotations on write*Snapshot()
     * methods to generate savepoints TODO Note: You should generate the savepoint based on the
     * release branch instead of the master.
     */
    private final MigrationVersion flinkGenerateSavepointVersion = null;

    private final MigrationVersion testMigrateVersion;

    public WindowOperatorMigrationTest(MigrationVersion testMigrateVersion) {
        this.testMigrateVersion = testMigrateVersion;
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeSessionWindowsWithCountTriggerSnapshot() throws Exception {
        final int sessionSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<String>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(new SessionWindowFunction()),
                                PurgingTrigger.of(CountTrigger.of(4)),
                                0,
                                null /* late data output tag */);

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        // add elements out-of-order
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 0));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 2), 1000));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 3), 2500));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 4), 3500));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 10));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 2), 1000));

        // do snapshot and save to file
        OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0L);

        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-session-with-stateful-trigger-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    @Test
    public void testRestoreSessionWindowsWithCountTrigger() throws Exception {

        final int sessionSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<String>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(new SessionWindowFunction()),
                                PurgingTrigger.of(CountTrigger.of(4)),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-session-with-stateful-trigger-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3), 2500));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 6000));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 2), 6500));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3), 7000));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple3ResultSortComparator());

        // add an element that merges the two "key1" sessions, they should now have count 6, and
        // therefore fire
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 10), 4500));

        expectedOutput.add(new StreamRecord<>(new Tuple3<>("key1-22", 10L, 10000L), 9999L));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple3ResultSortComparator());

        testHarness.close();
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeSessionWindowsWithCountTriggerInMintConditionSnapshot() throws Exception {

        final int sessionSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<String>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(new SessionWindowFunction()),
                                PurgingTrigger.of(CountTrigger.of(4)),
                                0,
                                null /* late data output tag */);

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        // do snapshot and save to file
        OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-session-with-stateful-trigger-mint-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    /**
     * This checks that we can restore from a virgin {@code WindowOperator} that has never seen any
     * elements.
     */
    @Test
    public void testRestoreSessionWindowsWithCountTriggerInMintCondition() throws Exception {

        final int sessionSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<String>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(new SessionWindowFunction()),
                                PurgingTrigger.of(CountTrigger.of(4)),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple3<String, Long, Long>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-session-with-stateful-trigger-mint-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        // add elements out-of-order
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 0));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 2), 1000));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 3), 2500));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 4), 3500));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 10));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 2), 1000));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3), 2500));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 6000));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 2), 6500));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3), 7000));

        expectedOutput.add(new StreamRecord<>(new Tuple3<>("key2-10", 0L, 6500L), 6499));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple3ResultSortComparator());

        // add an element that merges the two "key1" sessions, they should now have count 6, and
        // therefore fire
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 10), 4500));

        expectedOutput.add(new StreamRecord<>(new Tuple3<>("key1-22", 10L, 10000L), 9999L));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple3ResultSortComparator());

        testHarness.close();
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeReducingEventTimeWindowsSnapshot() throws Exception {
        final int windowSize = 3;

        ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ReducingStateDescriptor<>(
                        "window-contents",
                        new SumReducer<>(),
                        STRING_INT_TUPLE.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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalSingleValueWindowFunction<>(
                                        new PassThroughWindowFunction<
                                                String, TimeWindow, Tuple2<String, Integer>>()),
                                EventTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        // add elements out-of-order
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 3999));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 3000));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 20));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 0));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 999));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 1998));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 1999));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("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
        OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-reduce-event-time-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    @Test
    public void testRestoreReducingEventTimeWindows() throws Exception {
        final int windowSize = 3;

        ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ReducingStateDescriptor<>(
                        "window-contents",
                        new SumReducer<>(),
                        STRING_INT_TUPLE.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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalSingleValueWindowFunction<>(
                                        new PassThroughWindowFunction<
                                                String, TimeWindow, Tuple2<String, Integer>>()),
                                EventTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-reduce-event-time-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        testHarness.processWatermark(new Watermark(2999));
        expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 2999));
        expectedOutput.add(new StreamRecord<>(new Tuple2<>("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<>("key2", 2), 5999));
        expectedOutput.add(new Watermark(5999));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple2ResultSortComparator<>());
        testHarness.close();
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeApplyEventTimeWindowsSnapshot() throws Exception {
        final int windowSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(
                                        new RichSumReducer<TimeWindow>()),
                                EventTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        // add elements out-of-order
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 3999));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 3000));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 20));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 0));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 999));

        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 1998));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 1999));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("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
        OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-apply-event-time-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    @Test
    public void testRestoreApplyEventTimeWindows() throws Exception {
        final int windowSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(
                                        new RichSumReducer<TimeWindow>()),
                                EventTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-apply-event-time-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        testHarness.processWatermark(new Watermark(2999));
        expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 2999));
        expectedOutput.add(new StreamRecord<>(new Tuple2<>("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<>("key2", 2), 5999));
        expectedOutput.add(new Watermark(5999));

        TestHarnessUtil.assertOutputEqualsSorted(
                "Output was not correct.",
                expectedOutput,
                testHarness.getOutput(),
                new Tuple2ResultSortComparator<>());
        testHarness.close();
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeReducingProcessingTimeWindowsSnapshot() throws Exception {
        final int windowSize = 3;

        ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ReducingStateDescriptor<>(
                        "window-contents",
                        new SumReducer<>(),
                        STRING_INT_TUPLE.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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalSingleValueWindowFunction<>(
                                        new PassThroughWindowFunction<
                                                String, TimeWindow, Tuple2<String, Integer>>()),
                                ProcessingTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        testHarness.setProcessingTime(10);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1)));

        testHarness.setProcessingTime(3010);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key3", 1)));

        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<>());

        // do snapshot and save to file
        OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-reduce-processing-time-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    @Test
    public void testRestoreReducingProcessingTimeWindows() throws Exception {
        final int windowSize = 3;

        ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ReducingStateDescriptor<>(
                        "window-contents",
                        new SumReducer<>(),
                        STRING_INT_TUPLE.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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalSingleValueWindowFunction<>(
                                        new PassThroughWindowFunction<
                                                String, TimeWindow, Tuple2<String, Integer>>()),
                                ProcessingTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-reduce-processing-time-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        testHarness.setProcessingTime(3020);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 3)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3)));

        testHarness.setProcessingTime(6000);

        expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 5999));
        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<>());
        testHarness.close();
    }

    /** Manually run this to write binary snapshot data. */
    @Ignore
    @Test
    public void writeApplyProcessingTimeWindowsSnapshot() throws Exception {
        final int windowSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(
                                        new RichSumReducer<TimeWindow>()),
                                ProcessingTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();
        testHarness.open();

        testHarness.setProcessingTime(10);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1)));

        testHarness.setProcessingTime(3010);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key3", 1)));

        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<>());

        // do snapshot and save to file
        OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
        OperatorSnapshotUtil.writeStateHandle(
                snapshot,
                "src/test/resources/win-op-migration-test-apply-processing-time-flink"
                        + flinkGenerateSavepointVersion
                        + "-snapshot");

        testHarness.close();
    }

    @Test
    public void testRestoreApplyProcessingTimeWindows() throws Exception {
        final int windowSize = 3;

        ListStateDescriptor<Tuple2<String, Integer>> stateDesc =
                new ListStateDescriptor<>(
                        "window-contents",
                        STRING_INT_TUPLE.createSerializer(new ExecutionConfig()));

        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<>(),
                                BasicTypeInfo.STRING_TYPE_INFO.createSerializer(
                                        new ExecutionConfig()),
                                stateDesc,
                                new InternalIterableWindowFunction<>(
                                        new RichSumReducer<TimeWindow>()),
                                ProcessingTimeTrigger.create(),
                                0,
                                null /* late data output tag */);

        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>>
                testHarness =
                        new KeyedOneInputStreamOperatorTestHarness<>(
                                operator, new TupleKeySelector<>(), BasicTypeInfo.STRING_TYPE_INFO);

        testHarness.setup();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-apply-processing-time-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        testHarness.open();

        testHarness.setProcessingTime(3020);
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 3)));
        testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 3)));

        testHarness.setProcessingTime(6000);

        expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 5999));
        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<>());
        testHarness.close();
    }

    /** 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
        OperatorSubtaskState 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();

        testHarness.initializeState(
                OperatorSnapshotUtil.getResourceFilename(
                        "win-op-migration-test-kryo-serialized-key-flink"
                                + testMigrateVersion
                                + "-snapshot"));

        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 K getKey(Tuple2<K, Integer> value) throws Exception {
            return value.f0;
        }
    }

    @SuppressWarnings("unchecked")
    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<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());
                }
                int comparison = sr0.getValue().f0.compareTo(sr1.getValue().f0);
                if (comparison != 0) {
                    return comparison;
                } else {
                    return sr0.getValue().f1 - sr1.getValue().f1;
                }
            }
        }
    }

    @SuppressWarnings("unchecked")
    private static class Tuple3ResultSortComparator implements Comparator<Object> {
        @Override
        public int compare(Object o1, Object o2) {
            if (o1 instanceof Watermark || o2 instanceof Watermark) {
                return 0;
            } else {
                StreamRecord<Tuple3<String, Long, Long>> sr0 =
                        (StreamRecord<Tuple3<String, Long, Long>>) o1;
                StreamRecord<Tuple3<String, Long, Long>> sr1 =
                        (StreamRecord<Tuple3<String, Long, Long>>) o2;
                if (sr0.getTimestamp() != sr1.getTimestamp()) {
                    return (int) (sr0.getTimestamp() - sr1.getTimestamp());
                }
                int comparison = sr0.getValue().f0.compareTo(sr1.getValue().f0);
                if (comparison != 0) {
                    return comparison;
                } else {
                    comparison = (int) (sr0.getValue().f1 - sr1.getValue().f1);
                    if (comparison != 0) {
                        return comparison;
                    }
                    return (int) (sr0.getValue().f1 - sr1.getValue().f1);
                }
            }
        }
    }

    private static class SumReducer<K> implements ReduceFunction<Tuple2<K, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public Tuple2<K, Integer> reduce(Tuple2<K, Integer> value1, Tuple2<K, Integer> value2)
                throws Exception {
            return new Tuple2<>(value2.f0, value1.f1 + value2.f1);
        }
    }

    private static class RichSumReducer<W extends Window>
            extends RichWindowFunction<
                    Tuple2<String, Integer>, Tuple2<String, Integer>, String, W> {
        private static final long serialVersionUID = 1L;

        private boolean openCalled = false;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            openCalled = true;
        }

        @Override
        public void close() throws Exception {
            super.close();
        }

        @Override
        public void apply(
                String key,
                W window,
                Iterable<Tuple2<String, Integer>> input,
                Collector<Tuple2<String, Integer>> out)
                throws Exception {

            if (!openCalled) {
                fail("Open was not called");
            }
            int sum = 0;

            for (Tuple2<String, Integer> t : input) {
                sum += t.f1;
            }
            out.collect(new Tuple2<>(key, sum));
        }
    }

    private static class SessionWindowFunction
            implements WindowFunction<
                    Tuple2<String, Integer>, Tuple3<String, Long, Long>, String, TimeWindow> {
        private static final long serialVersionUID = 1L;

        @Override
        public void apply(
                String key,
                TimeWindow window,
                Iterable<Tuple2<String, Integer>> values,
                Collector<Tuple3<String, Long, Long>> out)
                throws Exception {
            int sum = 0;
            for (Tuple2<String, Integer> i : values) {
                sum += i.f1;
            }
            String resultString = key + "-" + sum;
            out.collect(new Tuple3<>(resultString, window.getStart(), window.getEnd()));
        }
    }
}