diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ClusterHealthSnapshot.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ClusterHealthSnapshot.java new file mode 100644 index 00000000000..cd5752ffc44 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/ClusterHealthSnapshot.java @@ -0,0 +1,80 @@ +/* + * 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.fluss.server.coordinator; + +import java.util.Collections; +import java.util.Map; + +/** + * An immutable, point-in-time view of cluster-wide and per-tablet-server replica/leader health, + * derived from {@link CoordinatorContext}. + * + *

{@code numReplicas}/{@code inSyncReplicas}/{@code numLeaderReplicas}/{@code + * activeLeaderReplicas} are the same cluster-wide aggregates {@code + * CoordinatorService#computeClusterHealth} reports; {@link #tabletServerLoads()} additionally + * attributes replicas/ISR membership/leadership to the specific server holding them. Both are + * computed together from a single pass over {@link CoordinatorContext#getAllBuckets()}. + * + *

Instances are published by {@link CoordinatorHealthCache} and are safe to read from any thread + * without synchronization: once constructed, an instance is never mutated. + */ +public final class ClusterHealthSnapshot { + + public static final ClusterHealthSnapshot EMPTY = + new ClusterHealthSnapshot(0, 0, 0, 0, Collections.emptyMap()); + + private final int numReplicas; + private final int inSyncReplicas; + private final int numLeaderReplicas; + private final int activeLeaderReplicas; + private final Map tabletServerLoads; + + ClusterHealthSnapshot( + int numReplicas, + int inSyncReplicas, + int numLeaderReplicas, + int activeLeaderReplicas, + Map tabletServerLoads) { + this.numReplicas = numReplicas; + this.inSyncReplicas = inSyncReplicas; + this.numLeaderReplicas = numLeaderReplicas; + this.activeLeaderReplicas = activeLeaderReplicas; + this.tabletServerLoads = Collections.unmodifiableMap(tabletServerLoads); + } + + public int numReplicas() { + return numReplicas; + } + + public int inSyncReplicas() { + return inSyncReplicas; + } + + public int numLeaderReplicas() { + return numLeaderReplicas; + } + + public int activeLeaderReplicas() { + return activeLeaderReplicas; + } + + /** Per-tablet-server load, keyed by server id. Includes live and shutting-down servers. */ + public Map tabletServerLoads() { + return tabletServerLoads; + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java index 96220fee120..4be825abd9a 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java @@ -188,6 +188,7 @@ public class CoordinatorEventProcessor implements EventProcessor { private final CoordinatorChangeWatcher coordinatorChangeWatcher; private final TabletServerChangeWatcher tabletServerChangeWatcher; private final CoordinatorMetadataCache serverMetadataCache; + private final CoordinatorHealthCache healthCache; private final CoordinatorRequestBatch coordinatorRequestBatch; private final String internalListenerName; private final CoordinatorMetricGroup coordinatorMetricGroup; @@ -218,7 +219,10 @@ public CoordinatorEventProcessor( this.coordinatorChannelManager = coordinatorChannelManager; this.coordinatorContext = coordinatorContext; this.replicaCapacityController = replicaCapacityController; - this.coordinatorEventManager = new CoordinatorEventManager(this, coordinatorMetricGroup); + this.healthCache = new CoordinatorHealthCache(); + this.coordinatorEventManager = + new CoordinatorEventManager( + this, coordinatorContext, healthCache, coordinatorMetricGroup); this.replicaStateMachine = new ReplicaStateMachine( coordinatorContext, @@ -300,6 +304,10 @@ public CoordinatorContext getCoordinatorContext() { return coordinatorContext; } + public CoordinatorHealthCache getHealthCache() { + return healthCache; + } + @VisibleForTesting TableLifecycleThrottler getLifecycleThrottler() { return lifecycleThrottler; @@ -501,6 +509,11 @@ private void initCoordinatorContext() throws Exception { tabletServerInfoList, coordinatorContext.getServerTags()); updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList); + // Warm the health cache once, unconditionally, so it isn't empty before the first + // relevant mutation lands. Bulk-loading buckets above does not mark it dirty on purpose + // (that would mean one dirty-mark per bucket during startup, for no benefit); this single + // explicit refresh covers it instead. + healthCache.refresh(coordinatorContext, true); // Auto-partition initialization schedules creation checks immediately. Start it only after // the observed KV leader replica count and live tablet server resources are restored. @@ -669,12 +682,16 @@ private void enqueueRetryOfflineLeaderEventSafely() { public void process(CoordinatorEvent event) { if (event instanceof CreateTableEvent) { processCreateTable((CreateTableEvent) event); + healthCache.onTopologyChanged(); } else if (event instanceof CreatePartitionEvent) { processCreatePartition((CreatePartitionEvent) event); + healthCache.onTopologyChanged(); } else if (event instanceof DropTableEvent) { processDropTable((DropTableEvent) event); + healthCache.onTopologyChanged(); } else if (event instanceof DropPartitionEvent) { processDropPartition((DropPartitionEvent) event); + healthCache.onTopologyChanged(); } else if (event instanceof SchemaChangeEvent) { SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event; processSchemaChange(schemaChangeEvent); @@ -693,8 +710,10 @@ public void process(CoordinatorEvent event) { processDeadCoordinator((DeadCoordinatorEvent) event); } else if (event instanceof NewTabletServerEvent) { processNewTabletServer((NewTabletServerEvent) event); + healthCache.onTabletServerRegistered(); } else if (event instanceof DeadTabletServerEvent) { processDeadTabletServer((DeadTabletServerEvent) event); + healthCache.onTabletServerDied(); } else if (event instanceof AdjustIsrReceivedEvent) { AdjustIsrReceivedEvent adjustIsrReceivedEvent = (AdjustIsrReceivedEvent) event; CompletableFuture callback = @@ -733,12 +752,21 @@ public void process(CoordinatorEvent event) { AddServerTagEvent addServerTagEvent = (AddServerTagEvent) event; completeFromCallable( addServerTagEvent.getRespCallback(), - () -> processAddServerTag(addServerTagEvent)); + () -> { + AddServerTagResponse response = processAddServerTag(addServerTagEvent); + healthCache.onTopologyChanged(); + return response; + }); } else if (event instanceof RemoveServerTagEvent) { RemoveServerTagEvent removeServerTagEvent = (RemoveServerTagEvent) event; completeFromCallable( removeServerTagEvent.getRespCallback(), - () -> processRemoveServerTag(removeServerTagEvent)); + () -> { + RemoveServerTagResponse response = + processRemoveServerTag(removeServerTagEvent); + healthCache.onTopologyChanged(); + return response; + }); } else if (event instanceof RebalanceEvent) { RebalanceEvent rebalanceEvent = (RebalanceEvent) event; completeFromCallable( @@ -1790,6 +1818,7 @@ private void onBucketReassignment( // A2. Set RS = TRS, AR = [], RR = [] in memory. coordinatorContext.updateBucketReplicaAssignment(tableBucket, reassignment.replicas); + healthCache.onTopologyChanged(); updateReplicaAssignmentForBucket(tableBucket, reassignment.replicas); // A3. replicas in AR -> NewReplica @@ -1815,6 +1844,7 @@ private void onBucketReassignment( maybeReassignedBucketLeaderIfRequired(tableBucket, targetReplicas); // B3. Set RS = TRS, AR = [], RR = [] in memory. coordinatorContext.updateBucketReplicaAssignment(tableBucket, targetReplicas); + healthCache.onTopologyChanged(); // B4. Re-send LeaderAndIsr request with new leader and a new RS (using TRS) and same // isr to every tabletServer in TRS. updateBucketEpochAndSendRequest(tableBucket, targetReplicas); @@ -1991,7 +2021,14 @@ private List tryProcessAdjustIsr( } // update coordinator leader and isr cache. - newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr); + newLeaderAndIsrList.forEach( + (tableBucket, newLeaderAndIsr) -> { + coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr); + healthCache.onBucketLeaderAndIsrChanged( + tableBucket, + coordinatorContext.getAssignment(tableBucket), + Optional.of(newLeaderAndIsr)); + }); // First, try to judge whether the bucket is in rebalance task when isr change. newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange); @@ -2552,6 +2589,10 @@ private void updateBucketEpochAndSendRequest(TableBucket tableBucket, ListThis mirrors the pattern {@code CoordinatorMetadataCache} already uses for server topology: a + * single {@code volatile} immutable {@link ClusterHealthSnapshot}, recomputed and swapped by the + * coordinator event thread, and read lock-free by any thread via {@link #getSnapshot()} without + * going through {@code AccessContextEvent}. The copy-on-write and coalescing mechanics themselves + * live in {@link CoalescingRefreshCache}, shared with (or reusable by) any other coordinator-local + * derived view that wants the same "callers report facts, this decides when to act" shape. + * + *

Callers report state transitions through the {@code onXxx} methods below; this class alone + * decides whether a transition is urgent (should be reflected almost immediately) or can be batched + * (reflected the next time the coordinator event queue drains). Callers never see or decide urgency + * themselves — they only report what changed. Recomputing the snapshot is still an {@code + * O(buckets)} scan, same as {@code CoordinatorService#computeClusterHealth} and {@code + * #computeTabletServerLoads} today; what changes is how often that scan runs and who waits for it, + * not its cost. + * + *

{@link #refresh(CoordinatorContext, boolean)} must only be called from the coordinator event + * thread, since {@code CoordinatorContext} is not thread-safe -- see {@link + * CoalescingRefreshCache}'s javadoc for the single-writer-thread assumption this relies on. + */ +public final class CoordinatorHealthCache { + + /** + * Upper bound on how long an urgent (degrading) change may sit unreflected while the event + * queue keeps draining new work. Bounds worst-case staleness for a safety-relevant signal + * without forcing a full rescan after every single event in a burst. + */ + @VisibleForTesting static final long URGENT_MAX_DELAY_MS = 200; + + private final CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(ClusterHealthSnapshot.EMPTY, URGENT_MAX_DELAY_MS); + + // -------------------------------------------------------------------------------------------- + // Reporting: callers state facts, this class decides what they mean. + // -------------------------------------------------------------------------------------------- + + /** + * Reports the current (post-mutation) leader/ISR state for a bucket. Under-replication or a + * missing leader is treated as urgent; anything else is batched. + */ + public void onBucketLeaderAndIsrChanged( + TableBucket tableBucket, List assignment, Optional current) { + boolean underReplicated = + current.map(lai -> lai.isr().size() < assignment.size()).orElse(true); + boolean leaderless = + current.map(lai -> lai.leader() == LeaderAndIsr.NO_LEADER).orElse(true); + cache.markDirty(underReplicated || leaderless); + } + + /** Reports that a bucket's leader became active or inactive. Inactive is urgent. */ + public void onLeaderActivityChanged(boolean isActive) { + cache.markDirty(!isActive); + } + + /** A tablet server died. Always urgent -- it can only make things worse. */ + public void onTabletServerDied() { + cache.markDirty(true); + } + + /** A tablet server registered (startup or rejoin). Never urgent. */ + public void onTabletServerRegistered() { + cache.markDirty(false); + } + + /** + * Catch-all for topology changes that don't represent degradation: table/partition + * create-delete, replica reassignment, server tag add/remove. + */ + public void onTopologyChanged() { + cache.markDirty(false); + } + + // -------------------------------------------------------------------------------------------- + // Refresh policy: the decision of when to act lives in CoalescingRefreshCache; the decision of + // what "urgent" means for this data (above) and how to compute a snapshot (below) live here. + // -------------------------------------------------------------------------------------------- + + /** + * Recomputes and republishes the snapshot if, and only if, something has actually changed since + * the last refresh and now is the right time to act on it: a non-urgent change needs {@code + * force}; an urgent change is bounded by {@link #URGENT_MAX_DELAY_MS} regardless of {@code + * force}. If nothing changed, this is a no-op regardless of {@code force} -- see {@link + * CoalescingRefreshCache}'s javadoc for why that check always comes first. + * + *

Must only be called from a thread that safely owns {@code ctx} (i.e. the coordinator event + * thread) — {@code ctx} itself is not thread-safe. + * + * @param force overrides the timing question directly. The coordinator event loop passes + * whether its own event queue is currently empty; an explicit warm-up (e.g. right after the + * coordinator finishes loading its initial state) or a test that wants to bypass the timing + * policy passes {@code true} unconditionally -- which still only takes effect because a + * freshly constructed cache starts dirty. + */ + public void refresh(CoordinatorContext ctx, boolean force) { + cache.refresh(() -> computeSnapshot(ctx), force); + } + + /** Returns the most recently published snapshot. Safe to call from any thread. */ + public ClusterHealthSnapshot getSnapshot() { + return cache.get(); + } + + @VisibleForTesting + boolean isDirty() { + return cache.isDirty(); + } + + @VisibleForTesting + boolean isUrgentlyDirty() { + return cache.isUrgentlyDirty(); + } + + /** + * Computes both the cluster-wide aggregates ({@code CoordinatorService#computeClusterHealth} + * semantics) and the per-server breakdown ({@code CoordinatorService#computeTabletServerLoads} + * semantics) from a single pass over {@code ctx.getAllBuckets()}, instead of one pass per view. + */ + private static ClusterHealthSnapshot computeSnapshot(CoordinatorContext ctx) { + Map loads = new TreeMap<>(); + // report live and shutting-down servers even if they host no replicas, so that + // an evacuated server explicitly shows zero replicas + for (int serverId : ctx.liveOrShuttingDownTabletServers()) { + getOrCreate(loads, serverId); + } + + int numReplicas = 0; + int inSyncReplicas = 0; + int numLeaderReplicas = 0; + int activeLeaderReplicas = 0; + + for (TableBucket tb : ctx.getAllBuckets()) { + List assignment = ctx.getAssignment(tb); + numReplicas += assignment.size(); + // matches CoordinatorService#computeClusterHealth: counts buckets, not leaders + numLeaderReplicas++; + for (int serverId : assignment) { + getOrCreate(loads, serverId).numReplicas++; + } + + boolean leaderActive = ctx.isLeaderActive(tb); + if (leaderActive) { + activeLeaderReplicas++; + } + + Optional laiOpt = ctx.getBucketLeaderAndIsr(tb); + if (laiOpt.isPresent()) { + LeaderAndIsr lai = laiOpt.get(); + inSyncReplicas += lai.isr().size(); + for (int serverId : lai.isr()) { + getOrCreate(loads, serverId).inSyncReplicas++; + } + if (lai.leader() != LeaderAndIsr.NO_LEADER) { + MutableLoad leaderLoad = getOrCreate(loads, lai.leader()); + leaderLoad.numLeaderReplicas++; + if (leaderActive) { + leaderLoad.activeLeaderReplicas++; + } + } + } + } + + Map tabletServerLoads = new HashMap<>(); + for (Map.Entry entry : loads.entrySet()) { + tabletServerLoads.put(entry.getKey(), entry.getValue().toImmutable()); + } + + return new ClusterHealthSnapshot( + numReplicas, + inSyncReplicas, + numLeaderReplicas, + activeLeaderReplicas, + tabletServerLoads); + } + + private static MutableLoad getOrCreate(Map loads, int serverId) { + return loads.computeIfAbsent(serverId, MutableLoad::new); + } + + /** Short-lived mutable accumulator used only while a snapshot is being computed. */ + private static final class MutableLoad { + private final int serverId; + private int numReplicas; + private int inSyncReplicas; + private int numLeaderReplicas; + private int activeLeaderReplicas; + + private MutableLoad(int serverId) { + this.serverId = serverId; + } + + private TabletServerLoad toImmutable() { + return new TabletServerLoad( + serverId, numReplicas, inSyncReplicas, numLeaderReplicas, activeLeaderReplicas); + } + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TabletServerLoad.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TabletServerLoad.java new file mode 100644 index 00000000000..d17db986719 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/TabletServerLoad.java @@ -0,0 +1,106 @@ +/* + * 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.fluss.server.coordinator; + +import java.util.Objects; + +/** + * An immutable, point-in-time replica/leader load for a single tablet server, derived from {@link + * CoordinatorContext}'s bucket assignment and leader/ISR state. + * + * @see ClusterHealthSnapshot + */ +public final class TabletServerLoad { + + private final int serverId; + private final int numReplicas; + private final int inSyncReplicas; + private final int numLeaderReplicas; + private final int activeLeaderReplicas; + + TabletServerLoad( + int serverId, + int numReplicas, + int inSyncReplicas, + int numLeaderReplicas, + int activeLeaderReplicas) { + this.serverId = serverId; + this.numReplicas = numReplicas; + this.inSyncReplicas = inSyncReplicas; + this.numLeaderReplicas = numLeaderReplicas; + this.activeLeaderReplicas = activeLeaderReplicas; + } + + public int serverId() { + return serverId; + } + + public int numReplicas() { + return numReplicas; + } + + public int inSyncReplicas() { + return inSyncReplicas; + } + + public int numLeaderReplicas() { + return numLeaderReplicas; + } + + public int activeLeaderReplicas() { + return activeLeaderReplicas; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof TabletServerLoad)) { + return false; + } + TabletServerLoad that = (TabletServerLoad) o; + return serverId == that.serverId + && numReplicas == that.numReplicas + && inSyncReplicas == that.inSyncReplicas + && numLeaderReplicas == that.numLeaderReplicas + && activeLeaderReplicas == that.activeLeaderReplicas; + } + + @Override + public int hashCode() { + return Objects.hash( + serverId, numReplicas, inSyncReplicas, numLeaderReplicas, activeLeaderReplicas); + } + + @Override + public String toString() { + return "TabletServerLoad{" + + "serverId=" + + serverId + + ", numReplicas=" + + numReplicas + + ", inSyncReplicas=" + + inSyncReplicas + + ", numLeaderReplicas=" + + numLeaderReplicas + + ", activeLeaderReplicas=" + + activeLeaderReplicas + + '}'; + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java index 3bf64f1ef3d..ef37a8ffae7 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java @@ -24,6 +24,7 @@ import org.apache.fluss.metrics.Histogram; import org.apache.fluss.metrics.MetricNames; import org.apache.fluss.server.coordinator.CoordinatorContext; +import org.apache.fluss.server.coordinator.CoordinatorHealthCache; import org.apache.fluss.server.coordinator.statemachine.ReplicaState; import org.apache.fluss.server.metrics.group.CoordinatorEventMetricGroup; import org.apache.fluss.server.metrics.group.CoordinatorMetricGroup; @@ -32,6 +33,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.util.Set; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; @@ -54,6 +57,8 @@ public final class CoordinatorEventManager implements EventManager { private final EventProcessor eventProcessor; private final CoordinatorMetricGroup coordinatorMetricGroup; + private final @Nullable CoordinatorContext coordinatorContext; + private final CoordinatorHealthCache healthCache; private final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); private final CoordinatorEventThread thread = @@ -84,7 +89,22 @@ public final class CoordinatorEventManager implements EventManager { public CoordinatorEventManager( EventProcessor eventProcessor, CoordinatorMetricGroup coordinatorMetricGroup) { + this(eventProcessor, null, new CoordinatorHealthCache(), coordinatorMetricGroup); + } + + /** + * @param coordinatorContext used to coalesce-refresh {@code healthCache} from the event + * thread's own loop; {@code null} disables that refresh entirely (used by callers, e.g. + * tests, that only care about the metrics-polling behavior of this class). + */ + public CoordinatorEventManager( + EventProcessor eventProcessor, + @Nullable CoordinatorContext coordinatorContext, + CoordinatorHealthCache healthCache, + CoordinatorMetricGroup coordinatorMetricGroup) { this.eventProcessor = eventProcessor; + this.coordinatorContext = coordinatorContext; + this.healthCache = healthCache; this.coordinatorMetricGroup = coordinatorMetricGroup; registerMetrics(); } @@ -263,6 +283,13 @@ public void doWork() throws Exception { lastMetricsUpdateTime = currentTime; } + // Coalesce health-cache refreshes the same way: at most once per drained queue, + // sooner only if healthCache itself decided a change was urgent. No AccessContextEvent + // needed -- this thread already owns coordinatorContext directly. + if (coordinatorContext != null) { + healthCache.refresh(coordinatorContext, queue.isEmpty()); + } + // Use poll with timeout instead of blocking take() so that the thread // wakes up periodically to update metrics even when no events arrive // (e.g., after coordinator restart with no client requests). diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/CoalescingRefreshCache.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/CoalescingRefreshCache.java new file mode 100644 index 00000000000..d4575b21518 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/CoalescingRefreshCache.java @@ -0,0 +1,131 @@ +/* + * 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.fluss.server.utils; + +import org.apache.fluss.annotation.VisibleForTesting; + +import javax.annotation.concurrent.GuardedBy; +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; + +import static org.apache.fluss.utils.concurrent.LockUtils.inLock; + +/** + * A copy-on-write cache of a derived, immutable value, recomputed on demand rather than on a fixed + * schedule, and coalesced so that many changes reported in a short window trigger at most one + * recompute. + * + *

Callers report that something changed via {@link #markDirty(boolean)}; whether that change was + * urgent affects only how soon {@link #refresh} is willing to act on it. The {@code force} flag on + * {@link #refresh} lets a caller override the timing question directly -- typically because it + * knows something this class doesn't, e.g. that its own work queue is currently empty, so a + * non-urgent change is now welcome to be acted on. An urgent change doesn't need the caller's help: + * it's bounded by {@code urgentMaxDelayMs} regardless of {@code force}, so a safety- or + * correctness-relevant change can't be starved by an owner that never passes {@code force=true}. + * + *

Crucially, {@code force} only ever overrides the timing gate, never the {@code dirty} + * gate: {@link #refresh} always checks "did anything actually change" first, unconditionally, + * before considering {@code force} at all. This is what keeps a quiet caller's cost at one boolean + * check per call -- if {@code force} skipped that check too, a caller that happens to be idle most + * of the time (the common case) would pay for a full recompute on every single call, whether or not + * anything had changed. A freshly constructed cache starts {@code dirty}, since it hasn't computed + * a real value yet -- that's what lets a one-time forced warm-up work without needing {@code force} + * to bypass the dirty check. + * + *

Either way, the eventual recompute is always a full recompute from the supplied {@link + * Supplier}, never incremental -- a wrong or missed {@code markDirty} call costs at worst one extra + * recompute, it never leaves the published value permanently wrong. + * + *

{@link #markDirty} and {@link #refresh} must both be called from a single owning thread; this + * class does no locking on that side, matching the assumption that whoever computes the new value + * also holds whatever non-thread-safe source that computation reads from. Only {@link #get()} is + * safe to call from any thread. + * + * @param the type of the derived, immutable snapshot value + */ +@ThreadSafe +public final class CoalescingRefreshCache { + + private final long urgentMaxDelayMs; + + private final Lock updateLock = new ReentrantLock(); + + @GuardedBy("updateLock") + private volatile T value; + + // only ever touched from the single owning thread -- see class javadoc. + // starts true: a freshly constructed cache hasn't computed a real value yet. + private boolean dirty = true; + private boolean urgentDirty; + private long lastRefreshTimeMs = System.currentTimeMillis(); + + public CoalescingRefreshCache(T initialValue, long urgentMaxDelayMs) { + this.value = initialValue; + this.urgentMaxDelayMs = urgentMaxDelayMs; + } + + /** Records that something changed. {@code urgent} affects only how soon it is acted on. */ + public void markDirty(boolean urgent) { + dirty = true; + if (urgent) { + urgentDirty = true; + } + } + + /** + * Recomputes and republishes the value via {@code compute}, if and only if something has + * actually changed since the last refresh and now is the right time to act on it. + * + * @param compute recomputes the value from scratch; only invoked if a refresh is due. + * @param force overrides the timing question directly -- pass {@code true} when the caller + * already knows now is a good time (e.g. its own work queue is empty), or to force an + * unconditional warm-up. Never overrides the {@code dirty} check: if nothing changed, this + * is a no-op regardless of {@code force}. + */ + public void refresh(Supplier compute, boolean force) { + if (!dirty) { + return; + } + long now = System.currentTimeMillis(); + boolean due = force || (urgentDirty && (now - lastRefreshTimeMs) >= urgentMaxDelayMs); + if (due) { + inLock(updateLock, () -> this.value = compute.get()); + lastRefreshTimeMs = now; + dirty = false; + urgentDirty = false; + } + } + + /** Returns the most recently published value. Safe to call from any thread. */ + public T get() { + return value; + } + + @VisibleForTesting + public boolean isDirty() { + return dirty; + } + + @VisibleForTesting + public boolean isUrgentlyDirty() { + return urgentDirty; + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHealthCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHealthCacheTest.java new file mode 100644 index 00000000000..ab25433aaae --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorHealthCacheTest.java @@ -0,0 +1,371 @@ +/* + * 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.fluss.server.coordinator; + +import org.apache.fluss.cluster.Endpoint; +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.rpc.messages.GetClusterHealthResponse; +import org.apache.fluss.server.metadata.ServerInfo; +import org.apache.fluss.server.zk.ZkEpoch; +import org.apache.fluss.server.zk.data.LeaderAndIsr; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link CoordinatorHealthCache}. */ +class CoordinatorHealthCacheTest { + + private CoordinatorContext ctx; + private CoordinatorHealthCache cache; + + @BeforeEach + void setUp() { + ctx = new CoordinatorContext(ZkEpoch.INITIAL_EPOCH); + ctx.setLiveTabletServers( + Arrays.asList(makeServerInfo(0), makeServerInfo(1), makeServerInfo(2))); + cache = new CoordinatorHealthCache(); + } + + @Test + void testInitialSnapshotIsEmpty() { + assertThat(cache.getSnapshot()).isSameAs(ClusterHealthSnapshot.EMPTY); + } + + @Test + void testEmptyClusterReportsZeroLoadPerLiveServer() { + cache.refresh(ctx, true); + + ClusterHealthSnapshot snapshot = cache.getSnapshot(); + assertThat(snapshot.numReplicas()).isZero(); + assertThat(snapshot.inSyncReplicas()).isZero(); + assertThat(snapshot.activeLeaderReplicas()).isZero(); + assertThat(snapshot.tabletServerLoads()).hasSize(3); + snapshot.tabletServerLoads() + .values() + .forEach( + load -> { + assertThat(load.numReplicas()).isZero(); + assertThat(load.inSyncReplicas()).isZero(); + assertThat(load.numLeaderReplicas()).isZero(); + assertThat(load.activeLeaderReplicas()).isZero(); + }); + } + + @Test + void testAggregatesMatchComputeClusterHealth() { + TableBucket tb1 = new TableBucket(1L, 0); + TableBucket tb2 = new TableBucket(1L, 1); + TableBucket tb3 = new TableBucket(2L, 0); + + ctx.updateBucketReplicaAssignment(tb1, Arrays.asList(0, 1)); + ctx.updateBucketReplicaAssignment(tb2, Arrays.asList(1, 2)); + ctx.updateBucketReplicaAssignment(tb3, Arrays.asList(0, 2)); + + ctx.putBucketLeaderAndIsr( + tb1, new LeaderAndIsr(0, 1, Arrays.asList(0, 1), Collections.emptyList(), 0, 1)); + ctx.putBucketLeaderAndIsr( + tb2, + new LeaderAndIsr( + 1, 1, Collections.singletonList(1), Collections.emptyList(), 0, 1)); + ctx.putBucketLeaderAndIsr( + tb3, new LeaderAndIsr(0, 1, Arrays.asList(0, 2), Collections.emptyList(), 0, 1)); + + GetClusterHealthResponse expected = CoordinatorService.computeClusterHealth(ctx); + + cache.refresh(ctx, true); + ClusterHealthSnapshot snapshot = cache.getSnapshot(); + + // the cache must reproduce the exact same cluster-wide aggregates as the + // AccessContextEvent-bound computation it is meant to replace. + assertThat(snapshot.numReplicas()).isEqualTo(expected.getNumReplicas()); + assertThat(snapshot.inSyncReplicas()).isEqualTo(expected.getInSyncReplicas()); + assertThat(snapshot.numLeaderReplicas()).isEqualTo(expected.getNumLeaderReplicas()); + assertThat(snapshot.activeLeaderReplicas()).isEqualTo(expected.getActiveLeaderReplicas()); + } + + @Test + void testPerServerBreakdownAttributesReplicasIsrAndLeader() { + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2)); + ctx.putBucketLeaderAndIsr( + tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1), Collections.emptyList(), 0, 1)); + + cache.refresh(ctx, true); + ClusterHealthSnapshot snapshot = cache.getSnapshot(); + + TabletServerLoad server0 = snapshot.tabletServerLoads().get(0); + TabletServerLoad server1 = snapshot.tabletServerLoads().get(1); + TabletServerLoad server2 = snapshot.tabletServerLoads().get(2); + + // server 0: replica + in ISR + is the leader (and active, since it is in the ISR) + assertThat(server0.numReplicas()).isEqualTo(1); + assertThat(server0.inSyncReplicas()).isEqualTo(1); + assertThat(server0.numLeaderReplicas()).isEqualTo(1); + assertThat(server0.activeLeaderReplicas()).isEqualTo(1); + + // server 1: replica + in ISR, not the leader + assertThat(server1.numReplicas()).isEqualTo(1); + assertThat(server1.inSyncReplicas()).isEqualTo(1); + assertThat(server1.numLeaderReplicas()).isZero(); + + // server 2: replica, but out of ISR (not in the isr list above) + assertThat(server2.numReplicas()).isEqualTo(1); + assertThat(server2.inSyncReplicas()).isZero(); + assertThat(server2.numLeaderReplicas()).isZero(); + + // sum of per-server replicas must reconcile with the cluster-wide aggregate + int sumOfServerReplicas = + snapshot.tabletServerLoads().values().stream() + .mapToInt(TabletServerLoad::numReplicas) + .sum(); + assertThat(sumOfServerReplicas).isEqualTo(snapshot.numReplicas()); + } + + @Test + void testUnattributedLeaderDoesNotReconcileWithBucketCountAggregate() { + // deliberately exercise the pre-existing semantic gap between + // CoordinatorService#computeClusterHealth's numLeaderReplicas (a bucket count, always + // incremented) and the per-server numLeaderReplicas (only incremented when a leader is + // actually assigned): with NO_LEADER, the aggregate still counts the bucket, but no + // server gets credited. + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1)); + ctx.putBucketLeaderAndIsr( + tb, + new LeaderAndIsr( + LeaderAndIsr.NO_LEADER, + 1, + Arrays.asList(0, 1), + Collections.emptyList(), + 0, + 1)); + + cache.refresh(ctx, true); + ClusterHealthSnapshot snapshot = cache.getSnapshot(); + + assertThat(snapshot.numLeaderReplicas()).isEqualTo(1); // one bucket, counted regardless + int sumOfPerServerLeaderReplicas = + snapshot.tabletServerLoads().values().stream() + .mapToInt(TabletServerLoad::numLeaderReplicas) + .sum(); + assertThat(sumOfPerServerLeaderReplicas).isZero(); // nobody is actually the leader + } + + @Test + void testEvacuatedLiveServerStillReportedWithZeroLoad() { + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Collections.singletonList(0)); + ctx.putBucketLeaderAndIsr( + tb, + new LeaderAndIsr( + 0, 1, Collections.singletonList(0), Collections.emptyList(), 0, 1)); + + cache.refresh(ctx, true); + + // server 1 and 2 host nothing, but are live, so they must still be present with 0 load + // (mirrors the "evacuated server explicitly shows zero replicas" contract). + TabletServerLoad server1 = cache.getSnapshot().tabletServerLoads().get(1); + assertThat(server1).isNotNull(); + assertThat(server1.numReplicas()).isZero(); + } + + @Test + void testPublishedSnapshotIsImmutableAcrossLaterUpdates() { + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1)); + cache.refresh(ctx, true); + + ClusterHealthSnapshot firstSnapshot = cache.getSnapshot(); + assertThat(firstSnapshot.numReplicas()).isEqualTo(2); + + // a later mutation + refresh must not retroactively change a snapshot a caller already + // holds a reference to -- that is the entire point of copy-on-write. + TableBucket tb2 = new TableBucket(2L, 0); + ctx.updateBucketReplicaAssignment(tb2, Arrays.asList(0, 1, 2)); + cache.onTopologyChanged(); // real callers always report through onXxx before refreshing + cache.refresh(ctx, true); + + assertThat(firstSnapshot.numReplicas()).isEqualTo(2); + assertThat(cache.getSnapshot().numReplicas()).isEqualTo(5); + assertThat(cache.getSnapshot()).isNotSameAs(firstSnapshot); + } + + @Test + void testConcurrentReadsNeverObserveATornSnapshot() throws InterruptedException { + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2)); + ctx.putBucketLeaderAndIsr( + tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1, 2), Collections.emptyList(), 0, 1)); + cache.refresh(ctx, true); + + AtomicBoolean stop = new AtomicBoolean(false); + AtomicReference failure = new AtomicReference<>(); + + Thread reader = + new Thread( + () -> { + while (!stop.get()) { + ClusterHealthSnapshot snapshot = cache.getSnapshot(); + int sumOfServerReplicas = + snapshot.tabletServerLoads().values().stream() + .mapToInt(TabletServerLoad::numReplicas) + .sum(); + // must ALWAYS reconcile: a torn read (fields from two different + // snapshot instances) would break this invariant. + if (sumOfServerReplicas != snapshot.numReplicas()) { + failure.set( + new AssertionError( + "torn snapshot: sumOfServerReplicas=" + + sumOfServerReplicas + + " numReplicas=" + + snapshot.numReplicas())); + return; + } + } + }); + + reader.start(); + // hammer refresh() from this thread while the reader spins, simulating the event thread + // republishing the snapshot concurrently with RPC-thread reads. onTopologyChanged() keeps + // marking it dirty so each iteration actually recomputes and swaps, not just the first. + for (int i = 0; i < 2000 && failure.get() == null; i++) { + cache.onTopologyChanged(); + cache.refresh(ctx, true); + } + stop.set(true); + reader.join(); + + assertThat(failure.get()).isNull(); + } + + @Test + void testFullIsrAndActiveLeaderIsNotUrgent() { + TableBucket tb = new TableBucket(1L, 0); + cache.onBucketLeaderAndIsrChanged( + tb, + Arrays.asList(0, 1, 2), + Optional.of( + new LeaderAndIsr( + 0, 1, Arrays.asList(0, 1, 2), Collections.emptyList(), 0, 1))); + + assertThat(cache.isDirty()).isTrue(); + assertThat(cache.isUrgentlyDirty()).isFalse(); + } + + @Test + void testUnderReplicatedIsrIsUrgent() { + TableBucket tb = new TableBucket(1L, 0); + cache.onBucketLeaderAndIsrChanged( + tb, + Arrays.asList(0, 1, 2), + Optional.of( + new LeaderAndIsr( + 0, 1, Arrays.asList(0, 1), Collections.emptyList(), 0, 1))); + + assertThat(cache.isUrgentlyDirty()).isTrue(); + } + + @Test + void testMissingLeaderIsUrgent() { + TableBucket tb = new TableBucket(1L, 0); + cache.onBucketLeaderAndIsrChanged(tb, Arrays.asList(0, 1), Optional.empty()); + + assertThat(cache.isUrgentlyDirty()).isTrue(); + } + + @Test + void testTabletServerDiedIsUrgentButRegisteredAndTopologyChangeAreNot() { + cache.onTabletServerDied(); + assertThat(cache.isUrgentlyDirty()).isTrue(); + + cache = new CoordinatorHealthCache(); + cache.onTabletServerRegistered(); + assertThat(cache.isDirty()).isTrue(); + assertThat(cache.isUrgentlyDirty()).isFalse(); + + cache = new CoordinatorHealthCache(); + cache.onTopologyChanged(); + assertThat(cache.isDirty()).isTrue(); + assertThat(cache.isUrgentlyDirty()).isFalse(); + + cache = new CoordinatorHealthCache(); + cache.onLeaderActivityChanged(false); + assertThat(cache.isUrgentlyDirty()).isTrue(); + } + + @Test + void testRefreshIsNoOpWhenNotDirty() { + cache.refresh(ctx, true); + ClusterHealthSnapshot warm = cache.getSnapshot(); + + // nothing reported dirty since the warm-up -- must not recompute, queue state aside. + cache.refresh(ctx, true); + assertThat(cache.getSnapshot()).isSameAs(warm); + } + + @Test + void testNonUrgentChangeWaitsForQueueToDrain() { + cache.refresh(ctx, true); + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1)); + cache.onTopologyChanged(); + + ClusterHealthSnapshot beforeDrain = cache.getSnapshot(); + cache.refresh(ctx, false); // queue still has work -- must not recompute yet + assertThat(cache.getSnapshot()).isSameAs(beforeDrain); + + cache.refresh(ctx, true); // queue drained -- now it should + assertThat(cache.getSnapshot()).isNotSameAs(beforeDrain); + assertThat(cache.getSnapshot().numReplicas()).isEqualTo(2); + } + + @Test + void testUrgentChangeIsBoundedByMaxDelayEvenIfQueueNeverDrains() throws InterruptedException { + cache.refresh(ctx, true); // establishes a real lastRefreshTimeMs baseline + TableBucket tb = new TableBucket(1L, 0); + ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1)); + cache.onTabletServerDied(); // urgent + + ClusterHealthSnapshot beforeDelay = cache.getSnapshot(); + cache.refresh(ctx, false); // queue busy, delay not yet elapsed -- must wait + assertThat(cache.getSnapshot()).isSameAs(beforeDelay); + + Thread.sleep(CoordinatorHealthCache.URGENT_MAX_DELAY_MS + 50); + + cache.refresh(ctx, false); // queue STILL busy, but the urgent bound is up + assertThat(cache.getSnapshot()).isNotSameAs(beforeDelay); + assertThat(cache.getSnapshot().numReplicas()).isEqualTo(2); + } + + private static ServerInfo makeServerInfo(int id) { + return new ServerInfo( + id, + "RACK" + id, + Endpoint.fromListenersString("CLIENT://host" + id + ":9124"), + ServerType.TABLET_SERVER); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/CoalescingRefreshCacheTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/CoalescingRefreshCacheTest.java new file mode 100644 index 00000000000..998971f3e9b --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/CoalescingRefreshCacheTest.java @@ -0,0 +1,139 @@ +/* + * 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.fluss.server.utils; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link CoalescingRefreshCache}, exercised with a trivial {@code Integer} value so the + * coalescing/urgency mechanics are verified independently of any real caller (e.g. {@code + * CoordinatorHealthCache}). + */ +class CoalescingRefreshCacheTest { + + private static final long URGENT_MAX_DELAY_MS = 100; + + @Test + void testFreshCacheStartsDirtyWithTheSeedValue() { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + assertThat(cache.get()).isEqualTo(0); + // hasn't computed a real value yet -- starts dirty so a warm-up can force it through. + assertThat(cache.isDirty()).isTrue(); + } + + @Test + void testRefreshIsNoOpWhenNotDirtyEvenIfForced() { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + cache.refresh(() -> 1, true); // clears the initial dirty state + AtomicInteger computeCalls = new AtomicInteger(); + + cache.refresh(() -> computeCalls.incrementAndGet(), true); + + // force overrides the timing gate, never the dirty gate -- nothing changed, so + // compute() must not even be invoked, force notwithstanding. + assertThat(computeCalls.get()).isZero(); + assertThat(cache.get()).isEqualTo(1); + } + + @Test + void testNonUrgentChangeWaitsForForce() { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + AtomicInteger source = new AtomicInteger(1); + cache.markDirty(false); + + cache.refresh(source::get, false); // not forced -- must wait + assertThat(cache.get()).isEqualTo(0); + assertThat(cache.isDirty()).isTrue(); + + cache.refresh(source::get, true); // forced -- now it should recompute + assertThat(cache.get()).isEqualTo(1); + assertThat(cache.isDirty()).isFalse(); + } + + @Test + void testUrgentChangeIsBoundedByMaxDelayEvenIfNeverForced() throws InterruptedException { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + AtomicInteger source = new AtomicInteger(1); + cache.refresh(source::get, true); // establishes a real lastRefreshTimeMs baseline + source.set(2); + + cache.markDirty(true); + cache.refresh(source::get, false); // urgent, but delay not yet elapsed + assertThat(cache.get()).isEqualTo(1); + + Thread.sleep(URGENT_MAX_DELAY_MS + 50); + + cache.refresh(source::get, false); // still not forced, but the bound is up + assertThat(cache.get()).isEqualTo(2); + assertThat(cache.isDirty()).isFalse(); + assertThat(cache.isUrgentlyDirty()).isFalse(); + } + + @Test + void testManyMarkDirtyCallsCoalesceIntoOneRecompute() { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + AtomicInteger computeCalls = new AtomicInteger(); + AtomicInteger source = new AtomicInteger(); + + for (int i = 0; i < 50; i++) { + cache.markDirty(false); + cache.refresh( + () -> { + computeCalls.incrementAndGet(); + return source.incrementAndGet(); + }, + false); // never forced -- simulates a burst arriving while the queue is busy + } + assertThat(computeCalls.get()).isZero(); + + cache.refresh( + () -> { + computeCalls.incrementAndGet(); + return source.incrementAndGet(); + }, + true); // forced now -- exactly one recompute for the whole burst + + assertThat(computeCalls.get()).isEqualTo(1); + assertThat(cache.get()).isEqualTo(1); + } + + @Test + void testForceClearsDirtyFlagsAfterRecomputing() { + CoalescingRefreshCache cache = + new CoalescingRefreshCache<>(0, URGENT_MAX_DELAY_MS); + cache.markDirty(true); + + cache.refresh(() -> 7, true); + + assertThat(cache.get()).isEqualTo(7); + // a forced refresh still recomputed the current truth, so nothing is left + // unreflected -- force participates fully in the dirty-tracking system, it doesn't + // sidestep it. + assertThat(cache.isDirty()).isFalse(); + assertThat(cache.isUrgentlyDirty()).isFalse(); + } +}