public void persist() throws Exception {
        if (!mapping.equals(initialMapping)) {
            state.update(
                    mapping.entrySet().stream()
                            .map((w) -> new Tuple2<>(w.getKey(), w.getValue()))
                            .collect(Collectors.toList()));
        }
    }
@Test
    void testPersist() throws Exception {
        @SuppressWarnings("unchecked")
        ListState<Tuple2<TimeWindow, TimeWindow>> mockState = mock(ListState.class);

        MergingWindowSet<TimeWindow> windowSet =
                new MergingWindowSet<>(
                        EventTimeSessionWindows.withGap(Time.milliseconds(3)), mockState);

        TestingMergeFunction mergeFunction = new TestingMergeFunction();

        windowSet.addWindow(new TimeWindow(1, 2), mergeFunction);
        windowSet.addWindow(new TimeWindow(17, 42), mergeFunction);

        assertThat(windowSet.getStateWindow(new TimeWindow(1, 2))).isEqualTo(new TimeWindow(1, 2));
        assertThat(windowSet.getStateWindow(new TimeWindow(17, 42)))
                .isEqualTo(new TimeWindow(17, 42));

        windowSet.persist();

        verify(mockState)
                .update(
                        eq(
                                new ArrayList<>(
                                        Arrays.asList(
                                                new Tuple2<>(
                                                        new TimeWindow(1, 2), new TimeWindow(1, 2)),
                                                new Tuple2<>(
                                                        new TimeWindow(17, 42),
                                                        new TimeWindow(17, 42))))));

        verify(mockState, times(1)).update(Matchers.anyObject());
    }