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?
---
