commit 3ef1c67d9ee46aab933e659a07c7d7dd4605a815
author mynameborat <bharath.kumarasubramanian@gmail.com> Wed Feb 19 13:42:08 2020 -0800
committer Bharath Kumarasubramanian <bharath.kumarasubramanian@gmail.com> Thu Feb 20 11:37:03 2020 -0800
tree 0c27906479e7ae8e312ff9ba0ebfc225b61b9b0a
parent f6aa8eeb312213a89a79396fe58d5d5b0c8802ca

SAMZA-2463: Duplicate firings of processing timers (#1282)

Remove the keys from the local book keeping map directly instead of using the keySet and removeAll

samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java

diff --git a/samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java b/samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java
index ddc5b29..cbebbde 100644
--- a/samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java
+++ b/samza-core/src/main/java/org/apache/samza/scheduler/EpochTimeScheduler.java
@@ -90,7 +90,11 @@
 
   public Map<TimerKey<?>, ScheduledCallback> removeReadyTimers() {
     final Map<TimerKey<?>, ScheduledCallback> timers = new TreeMap<>(readyTimers);
-    readyTimers.keySet().removeAll(timers.keySet());
+    // Remove keys on the map directly instead of using key set iterator and remove all
+    // on the key set as it results in duplicate firings due to weakly consistent SetView
+    for (TimerKey<?> key : timers.keySet()) {
+      readyTimers.remove(key);
+    }
     return timers;
   }