From ea5211af5a475f2b2a52612d49ba47ddf1afffbe Mon Sep 17 00:00:00 2001 From: Eliaazzz Date: Thu, 20 Aug 2026 10:50:41 +1000 Subject: [PATCH 1/2] [Spark] Fire processing-time timers in timestamp order SparkTimerInternals.getNextProcessingTimer() picked the eligible timer with the latest timestamp, so timers due in the same micro-batch fired in reverse order and a later callback observed state its predecessors had not written yet. It now returns the earliest one, and the expired timers fired by TimerUtils.triggerExpiredTimers are sorted the same way. Earliest-first also surfaces superseded settings of a re-set timer, so timers are now stored in a map keyed by namespace, id and family: a later setting replaces the prior one, per the TimerInternals contract, and timers restored from state that predates this change collapse to the setting with the latest target. --- CHANGES.md | 1 + .../spark/stateful/SparkTimerInternals.java | 56 +++++---- .../beam/runners/spark/util/TimerUtils.java | 7 +- .../stateful/SparkTimerInternalsTest.java | 107 ++++++++++++++++++ .../runners/spark/util/TimerUtilsTest.java | 21 ++++ 5 files changed, 167 insertions(+), 25 deletions(-) create mode 100644 runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java diff --git a/CHANGES.md b/CHANGES.md index 73966a48313c..350796bd00b8 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -80,6 +80,7 @@ ## Bugfixes +* (Java) Fixed the Spark runner firing processing-time timers in reverse timestamp order ([#39824](https://github.com/apache/beam/issues/39824)). * (Python) Fixed incorrect profiler options handling on portable runners ([#39613](https://github.com/apache/beam/issues/39613)). * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)). diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java index 9ef75635c212..cb363d326546 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java @@ -25,15 +25,15 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.apache.beam.runners.core.StateNamespace; import org.apache.beam.runners.core.TimerInternals; import org.apache.beam.runners.spark.coders.CoderHelpers; import org.apache.beam.runners.spark.util.GlobalWatermarkHolder.SparkWatermarks; import org.apache.beam.sdk.state.TimeDomain; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets; import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.Instant; @@ -44,7 +44,8 @@ public class SparkTimerInternals implements TimerInternals { private final Instant highWatermark; private final Instant synchronizedProcessingTime; - private final Set timers = Sets.newConcurrentHashSet(); + // Timers keyed by namespace, id and family, so a later setting replaces the prior one. + private final Map, TimerData> timers = new ConcurrentHashMap<>(); private Instant inputWatermark; @@ -105,37 +106,46 @@ public static SparkTimerInternals global(Map watermark } public Collection getTimers() { - return timers; + return timers.values(); } public void addTimers(Iterator timers) { while (timers.hasNext()) { TimerData timer = timers.next(); - this.timers.add(timer); + // State written before setTimer replaced prior settings can carry several settings of + // one timer; collapse them to the setting with the latest target. + this.timers.merge( + logicalKey(timer), + timer, + (existing, restored) -> + restored.getTimestamp().isAfter(existing.getTimestamp()) ? restored : existing); } } @Override public void setTimer(TimerData timer) { - this.timers.add(timer); + // A later setting of the same timer clears the prior one, per the TimerInternals contract. + this.timers.put(logicalKey(timer), timer); } @Override public void deleteTimer( StateNamespace namespace, String timerId, String timerFamilyId, TimeDomain timeDomain) { - this.timers.stream() - .filter( - timer -> - namespace.equals(timer.getNamespace()) - && timerId.equals(timer.getTimerId()) - && timerFamilyId.equals(timer.getTimerFamilyId()) - && timeDomain.equals(timer.getDomain())) - .forEach(this::deleteTimer); + List key = ImmutableList.of(namespace, timerId, timerFamilyId); + TimerData existing = this.timers.get(key); + if (existing != null && timeDomain.equals(existing.getDomain())) { + this.timers.remove(key, existing); + } } @Override public void deleteTimer(TimerData timer) { - this.timers.remove(timer); + // Deletes this setting only, so a setting made by the fired callback survives. + this.timers.remove(logicalKey(timer), timer); + } + + private static List logicalKey(TimerData timer) { + return ImmutableList.of(timer.getNamespace(), timer.getTimerId(), timer.getTimerFamilyId()); } @Override @@ -199,7 +209,7 @@ public static Iterator deserializeTimers( */ public boolean hasNextProcessingTimer() { final Instant currentProcessingTime = this.currentProcessingTime(); - return this.timers.stream() + return this.timers.values().stream() .anyMatch( (TimerData timerData) -> timerData.getDomain().equals(TimeDomain.PROCESSING_TIME) @@ -207,23 +217,23 @@ public boolean hasNextProcessingTimer() { } /** - * Finds the latest timer in {@link TimeDomain#PROCESSING_TIME} domain that has expired based on - * the current processing time. + * Finds the earliest timer in {@link TimeDomain#PROCESSING_TIME} domain that has expired based + * on the current processing time. * *

A timer is considered expired when its timestamp is less than the current processing time. - * If multiple expired timers exist, the one with the latest timestamp will be returned. + * Expired timers fire in timestamp order. * - * @return The expired processing timer with the latest timestamp if one exists, or {@code null} - * if no processing timers are ready to fire. + * @return The expired processing timer with the earliest timestamp if one exists, or {@code + * null} if no processing timers are ready to fire. */ public @Nullable TimerData getNextProcessingTimer() { final Instant currentProcessingTime = this.currentProcessingTime(); - return this.timers.stream() + return this.timers.values().stream() .filter( (TimerData timerData) -> timerData.getDomain().equals(TimeDomain.PROCESSING_TIME) && currentProcessingTime.isAfter(timerData.getTimestamp())) - .max(Comparator.comparing(TimerData::getTimestamp)) + .min(Comparator.comparing(TimerData::getTimestamp)) .orElse(null); } diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/util/TimerUtils.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/util/TimerUtils.java index 2111867d3851..6014873d6bfa 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/util/TimerUtils.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/util/TimerUtils.java @@ -21,6 +21,7 @@ import java.io.Serializable; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.Iterator; import java.util.List; @@ -202,10 +203,12 @@ public static void triggerExpiredTimers( SparkTimerInternals sparkTimerInternals, WindowingStrategy windowingStrategy, AbstractInOutIterator abstractInOutIterator) { - final Collection expiredTimers = + final List expiredTimers = getExpiredTimers(sparkTimerInternals, windowingStrategy); if (!expiredTimers.isEmpty()) { + // Timers fire in timestamp order. + expiredTimers.sort(Comparator.comparing(TimerInternals.TimerData::getTimestamp)); expiredTimers.forEach(abstractInOutIterator::fireTimer); } } @@ -221,7 +224,7 @@ public static void dropExpiredTimers( } } - private static Collection getExpiredTimers( + private static List getExpiredTimers( SparkTimerInternals sparkTimerInternals, WindowingStrategy windowingStrategy) { return sparkTimerInternals.getTimers().stream() .filter( diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java new file mode 100644 index 000000000000..f603e771b52a --- /dev/null +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.runners.spark.stateful; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import java.util.ArrayList; +import java.util.List; +import org.apache.beam.runners.core.StateNamespaces; +import org.apache.beam.runners.core.TimerInternals.TimerData; +import org.apache.beam.sdk.state.TimeDomain; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.joda.time.Instant; +import org.junit.Test; + +/** Tests for {@link SparkTimerInternals}. */ +public class SparkTimerInternalsTest { + + private static TimerData processingTimer(String timerId, Instant timestamp) { + return TimerData.of( + timerId, + "", + StateNamespaces.global(), + timestamp, + timestamp, + TimeDomain.PROCESSING_TIME); + } + + @Test + public void testProcessingTimersFireInTimestampOrder() { + SparkTimerInternals timerInternals = SparkTimerInternals.global(null); + + TimerData first = processingTimer("first", new Instant(1000)); + TimerData second = processingTimer("second", new Instant(2000)); + TimerData third = processingTimer("third", new Instant(3000)); + + // Set out of order; firing order must follow the timestamps. + timerInternals.setTimer(second); + timerInternals.setTimer(third); + timerInternals.setTimer(first); + + // Drain the way ParDoStateUpdateFn.SparkTimerInternalsIterator does. + List fired = new ArrayList<>(); + TimerData timer; + while ((timer = timerInternals.getNextProcessingTimer()) != null) { + fired.add(timer); + timerInternals.deleteTimer(timer); + } + + assertEquals(ImmutableList.of(first, second, third), fired); + } + + @Test + public void testSettingATimerAgainClearsThePriorSetting() { + SparkTimerInternals timerInternals = SparkTimerInternals.global(null); + + timerInternals.setTimer(processingTimer("timer", new Instant(1000))); + TimerData latest = processingTimer("timer", new Instant(2000)); + timerInternals.setTimer(latest); + + assertEquals(ImmutableList.of(latest), ImmutableList.copyOf(timerInternals.getTimers())); + assertEquals(latest, timerInternals.getNextProcessingTimer()); + } + + @Test + public void testAddTimersKeepsTheLatestSettingOfATimer() { + // State written before setTimer replaced prior settings can carry several settings of one + // timer; the setting with the latest target wins regardless of restore order. + TimerData earlier = processingTimer("timer", new Instant(1000)); + TimerData latest = processingTimer("timer", new Instant(2000)); + + SparkTimerInternals timerInternals = SparkTimerInternals.global(null); + timerInternals.addTimers(ImmutableList.of(earlier, latest).iterator()); + assertEquals(ImmutableList.of(latest), ImmutableList.copyOf(timerInternals.getTimers())); + + timerInternals = SparkTimerInternals.global(null); + timerInternals.addTimers(ImmutableList.of(latest, earlier).iterator()); + assertEquals(ImmutableList.of(latest), ImmutableList.copyOf(timerInternals.getTimers())); + } + + @Test + public void testGetNextProcessingTimerIgnoresEventTimeTimers() { + SparkTimerInternals timerInternals = SparkTimerInternals.global(null); + timerInternals.setTimer( + TimerData.of( + "event", "", StateNamespaces.global(), new Instant(0), new Instant(0), + TimeDomain.EVENT_TIME)); + + assertNull(timerInternals.getNextProcessingTimer()); + } +} diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java index a91b92aefadb..28f93d9ec24e 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java @@ -18,6 +18,8 @@ package org.apache.beam.runners.spark.util; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -36,6 +38,7 @@ import org.joda.time.Instant; import org.junit.Before; import org.junit.Test; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.MockitoAnnotations; @@ -111,6 +114,24 @@ public void testTriggerExpiredTimersWithEmptyTimers() { verify(mockIterator, never()).fireTimer(any()); } + @Test + public void testTriggerExpiredTimersFiresInTimestampOrder() { + // An even older expired timer, listed after the newer one. + TimerInternals.TimerData olderExpiredTimer = mock(TimerInternals.TimerData.class); + when(olderExpiredTimer.getTimestamp()) + .thenReturn(NOW.minus(ALLOWED_LATENESS.plus(Duration.standardMinutes(2)))); + when(olderExpiredTimer.getDomain()).thenReturn(TimeDomain.EVENT_TIME); + when(mockTimerInternals.getTimers()) + .thenReturn(Arrays.asList(expiredTimer, olderExpiredTimer)); + + TimerUtils.triggerExpiredTimers(mockTimerInternals, mockWindowingStrategy, mockIterator); + + // Expired timers fire in timestamp order. + InOrder inOrder = inOrder(mockIterator); + inOrder.verify(mockIterator).fireTimer(olderExpiredTimer); + inOrder.verify(mockIterator).fireTimer(expiredTimer); + } + @Test public void testTriggerExpiredTimersWithProcessingTimeDomain() { // Set up a processing-time timer From 026dd29649fed7bff1102d08fa00b860fc3974f9 Mon Sep 17 00:00:00 2001 From: Elia LIU Date: Fri, 21 Aug 2026 16:19:10 +1000 Subject: [PATCH 2/2] [Spark] Apply spotless formatting Co-Authored-By: Claude Opus 5 (1M context) --- .../runners/spark/stateful/SparkTimerInternals.java | 8 ++++---- .../spark/stateful/SparkTimerInternalsTest.java | 13 ++++++------- .../beam/runners/spark/util/TimerUtilsTest.java | 3 +-- 3 files changed, 11 insertions(+), 13 deletions(-) diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java index cb363d326546..141c58f1a23b 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java @@ -217,14 +217,14 @@ public boolean hasNextProcessingTimer() { } /** - * Finds the earliest timer in {@link TimeDomain#PROCESSING_TIME} domain that has expired based - * on the current processing time. + * Finds the earliest timer in {@link TimeDomain#PROCESSING_TIME} domain that has expired based on + * the current processing time. * *

A timer is considered expired when its timestamp is less than the current processing time. * Expired timers fire in timestamp order. * - * @return The expired processing timer with the earliest timestamp if one exists, or {@code - * null} if no processing timers are ready to fire. + * @return The expired processing timer with the earliest timestamp if one exists, or {@code null} + * if no processing timers are ready to fire. */ public @Nullable TimerData getNextProcessingTimer() { final Instant currentProcessingTime = this.currentProcessingTime(); diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java index f603e771b52a..9b65b683ff25 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/stateful/SparkTimerInternalsTest.java @@ -34,12 +34,7 @@ public class SparkTimerInternalsTest { private static TimerData processingTimer(String timerId, Instant timestamp) { return TimerData.of( - timerId, - "", - StateNamespaces.global(), - timestamp, - timestamp, - TimeDomain.PROCESSING_TIME); + timerId, "", StateNamespaces.global(), timestamp, timestamp, TimeDomain.PROCESSING_TIME); } @Test @@ -99,7 +94,11 @@ public void testGetNextProcessingTimerIgnoresEventTimeTimers() { SparkTimerInternals timerInternals = SparkTimerInternals.global(null); timerInternals.setTimer( TimerData.of( - "event", "", StateNamespaces.global(), new Instant(0), new Instant(0), + "event", + "", + StateNamespaces.global(), + new Instant(0), + new Instant(0), TimeDomain.EVENT_TIME)); assertNull(timerInternals.getNextProcessingTimer()); diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java index 28f93d9ec24e..24fbffd9e117 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/util/TimerUtilsTest.java @@ -121,8 +121,7 @@ public void testTriggerExpiredTimersFiresInTimestampOrder() { when(olderExpiredTimer.getTimestamp()) .thenReturn(NOW.minus(ALLOWED_LATENESS.plus(Duration.standardMinutes(2)))); when(olderExpiredTimer.getDomain()).thenReturn(TimeDomain.EVENT_TIME); - when(mockTimerInternals.getTimers()) - .thenReturn(Arrays.asList(expiredTimer, olderExpiredTimer)); + when(mockTimerInternals.getTimers()).thenReturn(Arrays.asList(expiredTimer, olderExpiredTimer)); TimerUtils.triggerExpiredTimers(mockTimerInternals, mockWindowingStrategy, mockIterator);