[
https://issues.apache.org/jira/browse/FLINK-9423?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16493611#comment-16493611
]
ASF GitHub Bot commented on FLINK-9423:
---------------------------------------
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6062#discussion_r191412983
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/HeapInternalTimerService.java
---
@@ -199,17 +186,9 @@ public long currentWatermark() {
@Override
public void registerProcessingTimeTimer(N namespace, long time) {
- InternalTimer<K, N> timer = new InternalTimer<>(time, (K)
keyContext.getCurrentKey(), namespace);
-
- // make sure we only put one timer per key into the queue
- Set<InternalTimer<K, N>> timerSet =
getProcessingTimeTimerSetForTimer(timer);
- if (timerSet.add(timer)) {
-
- InternalTimer<K, N> oldHead =
processingTimeTimersQueue.peek();
+ InternalTimer<K, N> oldHead = processingTimeTimersQueue.peek();
+ if (processingTimeTimersQueue.scheduleTimer(time, (K)
keyContext.getCurrentKey(), namespace)) {
--- End diff --
I know this is out of scope, but I think we could get rid of the
`KeyContext` by passing the current key to the `registerProcessingTimeTimer`
method. Moreover, instead of calling `KeyContext#setCurrentKey` we could pass
the key value to the `Triggerable#onEvent/ProcessingTime` method. Triggering
side effects via the `KeyContext` before calling certain methods is imo very
brittle.
> Implement efficient deletes for heap based timer service
> --------------------------------------------------------
>
> Key: FLINK-9423
> URL: https://issues.apache.org/jira/browse/FLINK-9423
> Project: Flink
> Issue Type: Improvement
> Components: Streaming
> Affects Versions: 1.5.0
> Reporter: Stefan Richter
> Assignee: Stefan Richter
> Priority: Major
> Fix For: 1.6.0
>
>
> The current data structures in the `HeapInternalTimerService` are not able to
> support efficient timer deletes, the complexity is currently O\(n\), where n
> is the number of registered timers.
>
> We can keep track of timer's positions in the priority queue and (in
> combination with the already existing set/map) have a more efficient
> algorithm for deletes.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
