Github user sihuazhou commented on a diff in the pull request: https://github.com/apache/flink/pull/6159#discussion_r195168815 --- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerHeap.java --- @@ -224,6 +227,15 @@ void bulkAddRestoredTimers(Collection<? extends InternalTimer<K, N>> restoredTim return result; } + @Nonnull + StateSnapshot snapshot(TimerHeapInternalTimer.TimerSerializer<K, N> serializer) { + return new InternalTimerHeapSnapshot<>( + Arrays.copyOfRange(queue, 1, size + 1), + serializer, --- End diff -- The `serializer` was be duplicated here or in the constructor of `InternalTimerHeapSnapshot` in the current code, do you plan to duplicate it when calling the `snapshot()` in the next code?
---