Merge pull request #5162 from vespa-engine/geirst/track-changes-in-cluster-stats-regarding-buckets-pending-in-global-space

Geirst/track changes in cluster stats regarding buckets pending in global space

17 files changed, 513 insertions, 82 deletions

diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java
new file mode 100644
index 00000000000..c9c1bbdbf79
--- /dev/null
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingChecker.java
@@ -0,0 +1,28 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+/**
+ * Class checking whether a particular bucket space on a content node might have buckets pending.
+ *
+ * Aggregated stats over the entire content cluster is used to check this.
+ */
+public class AggregatedStatsMergePendingChecker implements MergePendingChecker {
+
+    private final ContentClusterStats clusterStats;
+
+    public AggregatedStatsMergePendingChecker(ContentClusterStats clusterStats) {
+        this.clusterStats = clusterStats;
+    }
+
+    @Override
+    public boolean mayHaveMergesPending(String bucketSpace, int contentNodeIndex) {
+        ContentNodeStats nodeStats = clusterStats.getContentNode(contentNodeIndex);
+        if (nodeStats != null) {
+            ContentNodeStats.BucketSpaceStats bucketSpaceStats = nodeStats.getBucketSpace(bucketSpace);
+            if (bucketSpaceStats != null && bucketSpaceStats.mayHaveBucketsPending()) {
+                return true;
+            }
+        }
+        return false;
+    }
+}
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
index ea638010ab7..62cd158c759 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStateView.java
@@ -84,7 +84,7 @@ public class ClusterStateView {
 
     public ClusterState getClusterState() { return clusterState; }
 
-    public void handleUpdatedHostInfo(Map<Integer, String> hostnames, NodeInfo node, HostInfo hostInfo) {
+    public void handleUpdatedHostInfo(NodeInfo node, HostInfo hostInfo) {
         if ( ! node.isDistributor()) return;
 
         final int hostVersion;
@@ -112,6 +112,10 @@ public class ClusterStateView {
                 StorageNodeStatsBridge.generate(hostInfo.getDistributor()));
     }
 
+    public ClusterStatsAggregator getStatsAggregator() {
+        return statsAggregator;
+    }
+
     public String toString() {
         return clusterState.toString();
     }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
index 3f7cd129fc1..4a2ce1420de 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregator.java
@@ -1,9 +1,9 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.clustercontroller.core;
 
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
+import com.yahoo.document.FixedBucketSpaces;
+
+import java.util.*;
 
 /**
  * Class that stores content cluster stats (with bucket space stats per node) for
@@ -27,6 +27,7 @@ import java.util.Set;
 public class ClusterStatsAggregator {
 
     private final Set<Integer> distributors;
+    private final Set<Integer> nonUpdatedDistributors;
 
     // Maps the distributor node index to a map of content node index to the
     // content node's stats.
@@ -39,6 +40,7 @@ public class ClusterStatsAggregator {
 
     ClusterStatsAggregator(Set<Integer> distributors, Set<Integer> storageNodes) {
         this.distributors = distributors;
+        nonUpdatedDistributors = new HashSet<>(distributors);
         aggregatedStats = new ContentClusterStats(storageNodes);
     }
 
@@ -46,6 +48,24 @@ public class ClusterStatsAggregator {
         return aggregatedStats;
     }
 
+    boolean hasUpdatesFromAllDistributors() {
+        return nonUpdatedDistributors.isEmpty();
+    }
+
+    boolean mayHaveBucketsPendingInGlobalSpace() {
+        if (!hasUpdatesFromAllDistributors()) {
+            return true;
+        }
+        AggregatedStatsMergePendingChecker checker = new AggregatedStatsMergePendingChecker(aggregatedStats);
+        for (Iterator<ContentNodeStats> itr = aggregatedStats.iterator(); itr.hasNext(); ) {
+            ContentNodeStats stats = itr.next();
+            if (checker.mayHaveMergesPending(FixedBucketSpaces.globalSpace(), stats.getNodeIndex())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
      * Update the aggregator with the newest available stats from a distributor.
      */
@@ -53,6 +73,7 @@ public class ClusterStatsAggregator {
         if (!distributors.contains(distributorIndex)) {
             return;
         }
+        nonUpdatedDistributors.remove(distributorIndex);
         addStatsFromDistributor(distributorIndex, clusterStats);
     }
 
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java
new file mode 100644
index 00000000000..3ed6fc332ed
--- /dev/null
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTracker.java
@@ -0,0 +1,38 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+/**
+ * Class tracking whether we have changes in current and previous cluster stats.
+ *
+ * The cluster stats are considered changed if the current and previous stats differs in whether
+ * they may have buckets pending in the 'global' bucket space. This signals that the ClusterStateBundle should be recomputed.
+ */
+public class ClusterStatsChangeTracker {
+
+    private ClusterStatsAggregator aggregator;
+    private boolean prevMayHaveBucketsPending;
+
+    public ClusterStatsChangeTracker(ClusterStatsAggregator aggregator) {
+        this.aggregator = aggregator;
+        this.prevMayHaveBucketsPending = false;
+    }
+
+    public void syncBucketsPendingFlag() {
+        prevMayHaveBucketsPending = aggregator.mayHaveBucketsPendingInGlobalSpace();
+    }
+
+    public void updateAggregator(ClusterStatsAggregator newAggregator) {
+        syncBucketsPendingFlag();
+        aggregator = newAggregator;
+    }
+
+    public boolean statsHaveChanged() {
+        if (!aggregator.hasUpdatesFromAllDistributors()) {
+            return false;
+        }
+        if (prevMayHaveBucketsPending != aggregator.mayHaveBucketsPendingInGlobalSpace()) {
+            return true;
+        }
+        return false;
+    }
+}
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStats.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStats.java
index cefb3c3c31f..72a7b2e4bcf 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStats.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStats.java
@@ -14,25 +14,36 @@ public class ContentNodeStats {
     private Map<String, BucketSpaceStats> bucketSpaces = new HashMap<>();
 
     public static class BucketSpaceStats {
+        private int invalidCount;
         private long bucketsTotal;
         private long bucketsPending;
 
         private BucketSpaceStats() {
+            this.invalidCount = 1;
             this.bucketsTotal = 0;
             this.bucketsPending = 0;
         }
 
-        private BucketSpaceStats(long bucketsTotal, long bucketsPending) {
+        private BucketSpaceStats(long bucketsTotal, long bucketsPending, boolean invalid) {
+            this.invalidCount = (invalid ? 1 : 0);
             this.bucketsTotal = bucketsTotal;
             this.bucketsPending = bucketsPending;
         }
 
-        public static BucketSpaceStats empty() {
+        public static BucketSpaceStats invalid() {
             return new BucketSpaceStats();
         }
 
+        public static BucketSpaceStats invalid(long bucketsTotal, long bucketsPending) {
+            return new BucketSpaceStats(bucketsTotal, bucketsPending, true);
+        }
+
         public static BucketSpaceStats of(long bucketsTotal, long bucketsPending) {
-            return new BucketSpaceStats(bucketsTotal, bucketsPending);
+            return new BucketSpaceStats(bucketsTotal, bucketsPending, false);
+        }
+
+        public static BucketSpaceStats empty() {
+            return new BucketSpaceStats(0, 0, false);
         }
 
         public long getBucketsTotal() {
@@ -43,7 +54,16 @@ public class ContentNodeStats {
             return bucketsPending;
         }
 
+        public boolean mayHaveBucketsPending() {
+            return (bucketsPending > 0) || (invalidCount > 0);
+        }
+
+        public boolean valid() {
+            return invalidCount == 0;
+        }
+
         public void merge(BucketSpaceStats rhs, int factor) {
+            this.invalidCount += (factor * rhs.invalidCount);
             this.bucketsTotal += (factor * rhs.bucketsTotal);
             this.bucketsPending += (factor * rhs.bucketsPending);
         }
@@ -53,18 +73,19 @@ public class ContentNodeStats {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             BucketSpaceStats that = (BucketSpaceStats) o;
-            return bucketsTotal == that.bucketsTotal &&
+            return invalidCount == that.invalidCount &&
+                    bucketsTotal == that.bucketsTotal &&
                     bucketsPending == that.bucketsPending;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(bucketsTotal, bucketsPending);
+            return Objects.hash(invalidCount, bucketsTotal, bucketsPending);
         }
 
         @Override
         public String toString() {
-            return "{bucketsTotal=" + bucketsTotal + ", bucketsPending=" + bucketsPending + "}";
+            return "{bucketsTotal=" + bucketsTotal + ", bucketsPending=" + bucketsPending + ", invalidCount=" + invalidCount + "}";
         }
     }
 
@@ -76,7 +97,7 @@ public class ContentNodeStats {
                         BucketSpaceStats.of(stats.getBucketStats().getTotal(),
                                 stats.getBucketStats().getPending()));
             } else {
-                this.bucketSpaces.put(stats.getName(), BucketSpaceStats.empty());
+                this.bucketSpaces.put(stats.getName(), BucketSpaceStats.invalid());
             }
         }
     }
@@ -104,7 +125,7 @@ public class ContentNodeStats {
         for (Map.Entry<String, BucketSpaceStats> entry : stats.bucketSpaces.entrySet()) {
             BucketSpaceStats statsToUpdate = bucketSpaces.get(entry.getKey());
             if (statsToUpdate == null && factor == 1) {
-                statsToUpdate = new BucketSpaceStats();
+                statsToUpdate = BucketSpaceStats.empty();
                 bucketSpaces.put(entry.getKey(), statsToUpdate);
             }
             if (statsToUpdate != null) {
@@ -113,6 +134,10 @@ public class ContentNodeStats {
         }
     }
 
+    public BucketSpaceStats getBucketSpace(String bucketSpace) {
+        return bucketSpaces.get(bucketSpace);
+    }
+
     public Map<String, BucketSpaceStats> getBucketSpaces() {
         return bucketSpaces;
     }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
index 1059434aac3..e100fea780a 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/FleetController.java
@@ -325,7 +325,7 @@ public class FleetController implements NodeStateOrHostInfoChangeHandler, NodeAd
     @Override
     public void handleUpdatedHostInfo(NodeInfo nodeInfo, HostInfo newHostInfo) {
         verifyInControllerThread();
-        stateVersionTracker.handleUpdatedHostInfo(stateChangeHandler.getHostnames(), nodeInfo, newHostInfo);
+        stateVersionTracker.handleUpdatedHostInfo(nodeInfo, newHostInfo);
     }
 
     @Override
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MaintenanceWhenPendingGlobalMerges.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MaintenanceWhenPendingGlobalMerges.java
index 16401a79f08..2f4ca09b584 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MaintenanceWhenPendingGlobalMerges.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MaintenanceWhenPendingGlobalMerges.java
@@ -61,6 +61,6 @@ public class MaintenanceWhenPendingGlobalMerges implements ClusterStateDeriver {
     }
 
     private boolean hasMergesNotDone(String bucketSpace, int nodeIndex) {
-        return mergePendingChecker.hasMergesPending(bucketSpace, nodeIndex);
+        return mergePendingChecker.mayHaveMergesPending(bucketSpace, nodeIndex);
     }
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MergePendingChecker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MergePendingChecker.java
index e1b844c53de..fa4422ebf66 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MergePendingChecker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/MergePendingChecker.java
@@ -8,6 +8,6 @@ package com.yahoo.vespa.clustercontroller.core;
  */
 public interface MergePendingChecker {
 
-    boolean hasMergesPending(String bucketSpace, int contentNodeIndex);
+    boolean mayHaveMergesPending(String bucketSpace, int contentNodeIndex);
 
 }
diff --git a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
index fffe1c95124..7232b2f9346 100644
--- a/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
+++ b/clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/StateVersionTracker.java
@@ -7,7 +7,6 @@ import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Keeps track of the active cluster state and handles the transition edges between
@@ -33,12 +32,14 @@ public class StateVersionTracker {
     private ClusterStateBundle currentClusterState = latestCandidateState;
 
     private ClusterStateView clusterStateView;
+    private ClusterStatsChangeTracker clusterStatsChangeTracker;
 
     private final LinkedList<ClusterStateHistoryEntry> clusterStateHistory = new LinkedList<>();
     private int maxHistoryEntryCount = 50;
 
     StateVersionTracker() {
         clusterStateView = ClusterStateView.create(currentUnversionedState.getBaselineClusterState());
+        clusterStatsChangeTracker = new ClusterStatsChangeTracker(clusterStateView.getStatsAggregator());
     }
 
     void setVersionRetrievedFromZooKeeper(final int version) {
@@ -84,6 +85,7 @@ public class StateVersionTracker {
     public void updateLatestCandidateStateBundle(final ClusterStateBundle candidateBundle) {
         assert(latestCandidateState.getBaselineClusterState().getVersion() == 0);
         latestCandidateState = candidateBundle;
+        clusterStatsChangeTracker.syncBucketsPendingFlag();
     }
 
     /**
@@ -122,6 +124,7 @@ public class StateVersionTracker {
                 newStateBundle.getBaselineClusterState().getDistributionBitCount());
         // TODO should this take place in updateLatestCandidateStateBundle instead? I.e. does it require a consolidated state?
         clusterStateView = ClusterStateView.create(currentClusterState.getBaselineClusterState());
+        clusterStatsChangeTracker.updateAggregator(clusterStateView.getStatsAggregator());
     }
 
     private void recordCurrentStateInHistoryAtTime(final long currentTimeMs) {
@@ -132,13 +135,13 @@ public class StateVersionTracker {
         }
     }
 
-    void handleUpdatedHostInfo(final Map<Integer, String> hostnames, final NodeInfo node, final HostInfo hostInfo) {
+    void handleUpdatedHostInfo(final NodeInfo node, final HostInfo hostInfo) {
         // TODO the wiring here isn't unit tested. Need mockable integration points.
-        clusterStateView.handleUpdatedHostInfo(hostnames, node, hostInfo);
+        clusterStateView.handleUpdatedHostInfo(node, hostInfo);
     }
 
     boolean bucketSpaceMergeCompletionStateHasChanged() {
-        return false; // TODO wire changes in merge info
+        return clusterStatsChangeTracker.statsHaveChanged();
     }
 
     /*
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java
new file mode 100644
index 00000000000..e5df589f8fe
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/AggregatedStatsMergePendingCheckerTest.java
@@ -0,0 +1,65 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class AggregatedStatsMergePendingCheckerTest {
+
+    private static class Fixture {
+
+        private final AggregatedStatsMergePendingChecker checker;
+
+        public Fixture(ContentClusterStatsBuilder builder) {
+            this.checker = new AggregatedStatsMergePendingChecker(builder.build());
+        }
+
+        public static Fixture fromBucketStats(long bucketsPending) {
+            return new Fixture(new ContentClusterStatsBuilder()
+                    .add(1, "default", 5, bucketsPending));
+        }
+
+        public static Fixture fromInvalidBucketStats() {
+            return new Fixture(new ContentClusterStatsBuilder()
+                    .add(1, "default"));
+        }
+
+        public boolean mayHaveMergesPending(String bucketSpace, int contentNodeIndex) {
+            return checker.mayHaveMergesPending(bucketSpace, contentNodeIndex);
+        }
+
+    }
+
+    @Test
+    public void unknown_content_node_has_no_merges_pending() {
+        Fixture f = Fixture.fromBucketStats(1);
+        assertFalse(f.mayHaveMergesPending("default", 2));
+    }
+
+    @Test
+    public void unknown_bucket_space_has_no_merges_pending() {
+        Fixture f = Fixture.fromBucketStats(1);
+        assertFalse(f.mayHaveMergesPending("global", 1));
+    }
+
+    @Test
+    public void valid_bucket_space_stats_can_have_no_merges_pending() {
+        Fixture f = Fixture.fromBucketStats(0);
+        assertFalse(f.mayHaveMergesPending("default", 1));
+    }
+
+    @Test
+    public void valid_bucket_space_stats_can_have_merges_pending() {
+        Fixture f = Fixture.fromBucketStats(1);
+        assertTrue(f.mayHaveMergesPending("default", 1));
+    }
+
+    @Test
+    public void invalid_bucket_space_stats_has_merges_pending() {
+        Fixture f = Fixture.fromInvalidBucketStats();
+        assertTrue(f.mayHaveMergesPending("default", 1));
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateViewTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateViewTest.java
index dc8a4a0d441..b456965d549 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateViewTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStateViewTest.java
@@ -6,8 +6,6 @@ import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import com.yahoo.vespa.clustercontroller.core.hostinfo.StorageNodeStatsBridge;
 import org.junit.Test;
 
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
@@ -18,7 +16,6 @@ import static org.mockito.Mockito.*;
  * @since 5.34
  */
 public class ClusterStateViewTest {
-    final Map<Integer, String> hostnames = new HashMap<>();
     final NodeInfo nodeInfo = mock(NodeInfo.class);
     final Node node = mock(Node.class);
     final ClusterStatsAggregator statsAggregator = mock(ClusterStatsAggregator.class);
@@ -33,7 +30,7 @@ public class ClusterStateViewTest {
     public void testWrongNodeType() {
         when(nodeInfo.isDistributor()).thenReturn(false);
 
-        clusterStateView.handleUpdatedHostInfo(hostnames, nodeInfo, createHostInfo("101"));
+        clusterStateView.handleUpdatedHostInfo(nodeInfo, createHostInfo("101"));
 
         verify(statsAggregator, never()).updateForDistributor(anyInt(), any());
     }
@@ -45,7 +42,7 @@ public class ClusterStateViewTest {
         when(nodeInfo.isDistributor()).thenReturn(true);
         when(clusterState.getVersion()).thenReturn(101);
 
-        clusterStateView.handleUpdatedHostInfo(hostnames, nodeInfo, createHostInfo("22"));
+        clusterStateView.handleUpdatedHostInfo(nodeInfo, createHostInfo("22"));
 
         verify(statsAggregator, never()).updateForDistributor(anyInt(), any());
     }
@@ -55,7 +52,7 @@ public class ClusterStateViewTest {
         when(nodeInfo.isDistributor()).thenReturn(true);
         when(clusterState.getVersion()).thenReturn(101);
 
-        clusterStateView.handleUpdatedHostInfo(hostnames, nodeInfo, createHostInfo("22"));
+        clusterStateView.handleUpdatedHostInfo(nodeInfo, createHostInfo("22"));
 
         verify(statsAggregator, never()).updateForDistributor(anyInt(), any());
     }
@@ -77,7 +74,7 @@ public class ClusterStateViewTest {
         when(nodeInfo.getNodeIndex()).thenReturn(3);
         when(clusterState.getVersion()).thenReturn(101);
 
-        clusterStateView.handleUpdatedHostInfo(hostnames, nodeInfo, hostInfo);
+        clusterStateView.handleUpdatedHostInfo(nodeInfo, hostInfo);
 
         verify(statsAggregator).updateForDistributor(3, StorageNodeStatsBridge.generate(hostInfo.getDistributor()));
     }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
index c92d414aac8..ac6417d0077 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsAggregatorTest.java
@@ -7,6 +7,8 @@ import org.junit.Test;
 import java.util.*;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author hakonhall
@@ -14,48 +16,24 @@ import static org.junit.Assert.assertEquals;
  */
 public class ClusterStatsAggregatorTest {
 
-    private static class StatsBuilder {
-        private final Map<Integer, Map<String, ContentNodeStats.BucketSpaceStats> > stats = new HashMap<>();
-
-        public StatsBuilder add(int nodeIndex, String bucketSpace, long bucketsTotal, long bucketsPending) {
-            return add(nodeIndex, bucketSpace, ContentNodeStats.BucketSpaceStats.of(bucketsTotal, bucketsPending));
-        }
-        public StatsBuilder add(int nodeIndex, String bucketSpace) {
-            return add(nodeIndex, bucketSpace, ContentNodeStats.BucketSpaceStats.empty());
-        }
-        public StatsBuilder add(int nodeIndex, String bucketSpace, ContentNodeStats.BucketSpaceStats bucketSpaceStats) {
-            Map<String, ContentNodeStats.BucketSpaceStats> contentNodeStats = stats.get(nodeIndex);
-            if (contentNodeStats == null) {
-                contentNodeStats = new HashMap<>();
-                stats.put(nodeIndex, contentNodeStats);
-            }
-            contentNodeStats.put(bucketSpace, bucketSpaceStats);
-            return this;
-        }
-        public StatsBuilder add(int nodeIndex) {
-            stats.put(nodeIndex, new HashMap<>());
-            return this;
-        }
-        public ContentClusterStats build() {
-            Map<Integer, ContentNodeStats> nodeToStatsMap = new HashMap<>();
-            stats.forEach((nodeIndex, bucketSpaces) ->
-                    nodeToStatsMap.put(nodeIndex, new ContentNodeStats(nodeIndex, bucketSpaces)));
-            return new ContentClusterStats(nodeToStatsMap);
-        }
-    }
-
     private static class Fixture {
         private ClusterStatsAggregator aggregator;
         public Fixture(Set<Integer> distributorNodes,
                 Set<Integer> contentNodes) {
             aggregator = new ClusterStatsAggregator(distributorNodes, contentNodes);
         }
-        public void update(int distributorIndex, StatsBuilder clusterStats) {
+        public void update(int distributorIndex, ContentClusterStatsBuilder clusterStats) {
             aggregator.updateForDistributor(distributorIndex, clusterStats.build());
         }
-        public void verify(StatsBuilder expectedStats) {
+        public void verify(ContentClusterStatsBuilder expectedStats) {
             assertEquals(expectedStats.build(), aggregator.getAggregatedStats());
         }
+        public boolean hasUpdatesFromAllDistributors() {
+            return aggregator.hasUpdatesFromAllDistributors();
+        }
+        public boolean mayHaveBucketsPendingInGlobalSpace() {
+            return aggregator.mayHaveBucketsPendingInGlobalSpace();
+        }
     }
 
     private static Set<Integer> distributorNodes(Integer... indices) {
@@ -69,7 +47,7 @@ public class ClusterStatsAggregatorTest {
     @Test
     public void aggregator_handles_updates_to_single_distributor_and_content_node() {
         Fixture f = new Fixture(distributorNodes(1), contentNodes(3));
-        StatsBuilder stats = new StatsBuilder()
+        ContentClusterStatsBuilder stats = new ContentClusterStatsBuilder()
                 .add(3, "default", 10, 1)
                 .add(3, "global", 11, 2);
         f.update(1, stats);
@@ -80,17 +58,17 @@ public class ClusterStatsAggregatorTest {
     public void aggregator_handles_updates_to_multiple_distributors_and_content_nodes() {
         Fixture f = new Fixture(distributorNodes(1, 2), contentNodes(3, 4));
 
-        f.update(1, new StatsBuilder()
+        f.update(1, new ContentClusterStatsBuilder()
                 .add(3, "default", 10, 1)
                 .add(3, "global", 11, 2)
                 .add(4, "default", 12, 3)
                 .add(4, "global", 13, 4));
-        f.update(2, new StatsBuilder()
+        f.update(2, new ContentClusterStatsBuilder()
                 .add(3, "default", 14, 5)
                 .add(3, "global", 15, 6)
                 .add(4, "default", 16, 7)
                 .add(4, "global", 17, 8));
-        f.verify(new StatsBuilder()
+        f.verify(new ContentClusterStatsBuilder()
                 .add(3, "default", 10 + 14, 1 + 5)
                 .add(3, "global", 11 + 15, 2 + 6)
                 .add(4, "default", 12 + 16, 3 + 7)
@@ -101,29 +79,29 @@ public class ClusterStatsAggregatorTest {
     public void aggregator_handles_multiple_updates_from_same_distributor() {
         Fixture f = new Fixture(distributorNodes(1, 2), contentNodes(3));
 
-        f.update(1, new StatsBuilder().add(3, "default"));
-        f.verify(new StatsBuilder().add(3, "default"));
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default"));
+        f.verify(new ContentClusterStatsBuilder().add(3, "default"));
 
-        f.update(2, new StatsBuilder().add(3, "default", 10, 1));
-        f.verify(new StatsBuilder().add(3, "default", 10, 1));
+        f.update(2, new ContentClusterStatsBuilder().add(3, "default", 10, 1));
+        f.verify(new ContentClusterStatsBuilder().addInvalid(3, "default", 10, 1));
 
-        f.update(1, new StatsBuilder().add(3, "default", 11, 2));
-        f.verify(new StatsBuilder().add(3, "default", 10 + 11, 1 + 2));
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default", 11, 2));
+        f.verify(new ContentClusterStatsBuilder().add(3, "default", 10 + 11, 1 + 2));
 
-        f.update(2, new StatsBuilder().add(3, "default", 15, 6));
-        f.verify(new StatsBuilder().add(3, "default", 11 + 15, 2 + 6));
+        f.update(2, new ContentClusterStatsBuilder().add(3, "default", 15, 6));
+        f.verify(new ContentClusterStatsBuilder().add(3, "default", 11 + 15, 2 + 6));
 
-        f.update(1, new StatsBuilder().add(3, "default", 16, 7));
-        f.verify(new StatsBuilder().add(3, "default", 15 + 16, 6 + 7));
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default", 16, 7));
+        f.verify(new ContentClusterStatsBuilder().add(3, "default", 15 + 16, 6 + 7));
 
-        f.update(2, new StatsBuilder().add(3, "default", 12, 3));
-        f.verify(new StatsBuilder().add(3, "default", 16 + 12, 7 + 3));
+        f.update(2, new ContentClusterStatsBuilder().add(3, "default", 12, 3));
+        f.verify(new ContentClusterStatsBuilder().add(3, "default", 16 + 12, 7 + 3));
     }
 
     @Test
     public void aggregator_handles_more_content_nodes_that_distributors() {
         Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
-        StatsBuilder stats = new StatsBuilder()
+        ContentClusterStatsBuilder stats = new ContentClusterStatsBuilder()
                 .add(3, "default", 10, 1)
                 .add(4, "default", 11, 2);
         f.update(1, stats);
@@ -134,9 +112,46 @@ public class ClusterStatsAggregatorTest {
     public void aggregator_ignores_updates_to_unknown_distributor() {
         Fixture f = new Fixture(distributorNodes(1), contentNodes(3));
         final int downDistributorIndex = 2;
-        f.update(downDistributorIndex, new StatsBuilder()
+        f.update(downDistributorIndex, new ContentClusterStatsBuilder()
                 .add(3, "default", 7, 3));
-        f.verify(new StatsBuilder().add(3));
+        f.verify(new ContentClusterStatsBuilder().add(3));
+    }
+
+    @Test
+    public void aggregator_tracks_when_it_has_updates_from_all_distributors() {
+        Fixture f = new Fixture(distributorNodes(1, 2), contentNodes(3));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default"));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(1, new ContentClusterStatsBuilder().add(3, "default", 10, 1));
+        assertFalse(f.hasUpdatesFromAllDistributors());
+        f.update(2, new ContentClusterStatsBuilder().add(3, "default"));
+        assertTrue(f.hasUpdatesFromAllDistributors());
+    }
+
+    @Test
+    public void cluster_without_updates_from_all_distributors_may_have_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        assertTrue(f.mayHaveBucketsPendingInGlobalSpace());
+    }
+
+    @Test
+    public void cluster_may_have_buckets_pending_in_global_space_if_one_node_has_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        f.update(1, new ContentClusterStatsBuilder()
+                .add(3, "global", 10, 0)
+                .add(4, "global", 11, 1));
+        assertTrue(f.mayHaveBucketsPendingInGlobalSpace());
+    }
+
+    @Test
+    public void cluster_does_not_have_buckets_pending_in_global_space_if_no_nodes_have_buckets_pending() {
+        Fixture f = new Fixture(distributorNodes(1), contentNodes(3, 4));
+        f.update(1, new ContentClusterStatsBuilder()
+                .add(3, "global", 10, 0)
+                .add(4, "global", 11, 0)
+                .add(4, "default", 12, 1));
+        assertFalse(f.mayHaveBucketsPendingInGlobalSpace());
     }
 
 }
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java
new file mode 100644
index 00000000000..8670c512f14
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ClusterStatsChangeTrackerTest.java
@@ -0,0 +1,68 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import com.google.common.collect.Sets;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ClusterStatsChangeTrackerTest {
+
+    private static class Fixture {
+        private ClusterStatsAggregator aggregator;
+        private ClusterStatsChangeTracker tracker;
+
+        public Fixture() {
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
+            tracker = new ClusterStatsChangeTracker(aggregator);
+        }
+
+        public void setBucketsPendingStats() {
+            updateStats(1);
+        }
+
+        public void setInSyncStats() {
+            updateStats(0);
+        }
+
+        public void updateStats(long bucketsPending) {
+            aggregator.updateForDistributor(1, new ContentClusterStatsBuilder()
+                    .add(2, "global", 5, bucketsPending).build());
+        }
+
+        public void updateAggregator() {
+            aggregator = new ClusterStatsAggregator(Sets.newHashSet(1), Sets.newHashSet(2));
+            tracker.updateAggregator(aggregator);
+        }
+
+        public boolean statsHaveChanged() {
+            return tracker.statsHaveChanged();
+        }
+
+    }
+
+    @Test
+    public void stats_have_not_changed_if_not_all_distributors_are_updated() {
+        Fixture f = new Fixture();
+        assertFalse(f.statsHaveChanged());
+    }
+
+    @Test
+    public void stats_have_changed_if_previous_buckets_pending_stats_are_different_from_current() {
+        Fixture f = new Fixture();
+
+        f.setInSyncStats();
+        assertFalse(f.statsHaveChanged());
+        f.setBucketsPendingStats();
+        assertTrue(f.statsHaveChanged());
+
+        f.updateAggregator(); // previous stats may now have buckets pending
+
+        f.setInSyncStats();
+        assertTrue(f.statsHaveChanged());
+        f.setBucketsPendingStats();
+        assertFalse(f.statsHaveChanged());
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentClusterStatsBuilder.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentClusterStatsBuilder.java
new file mode 100644
index 00000000000..16767cafa8f
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentClusterStatsBuilder.java
@@ -0,0 +1,47 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Builder used for testing only.
+ */
+public class ContentClusterStatsBuilder {
+
+    private final Map<Integer, Map<String, ContentNodeStats.BucketSpaceStats>> stats = new HashMap<>();
+
+    public ContentClusterStatsBuilder add(int nodeIndex, String bucketSpace, long bucketsTotal, long bucketsPending) {
+        return add(nodeIndex, bucketSpace, ContentNodeStats.BucketSpaceStats.of(bucketsTotal, bucketsPending));
+    }
+
+    public ContentClusterStatsBuilder addInvalid(int nodeIndex, String bucketSpace, long bucketsTotal, long bucketsPending) {
+        return add(nodeIndex, bucketSpace, ContentNodeStats.BucketSpaceStats.invalid(bucketsTotal, bucketsPending));
+    }
+
+    public ContentClusterStatsBuilder add(int nodeIndex, String bucketSpace) {
+        return add(nodeIndex, bucketSpace, ContentNodeStats.BucketSpaceStats.invalid());
+    }
+
+    public ContentClusterStatsBuilder add(int nodeIndex, String bucketSpace, ContentNodeStats.BucketSpaceStats bucketSpaceStats) {
+        Map<String, ContentNodeStats.BucketSpaceStats> contentNodeStats = stats.get(nodeIndex);
+        if (contentNodeStats == null) {
+            contentNodeStats = new HashMap<>();
+            stats.put(nodeIndex, contentNodeStats);
+        }
+        contentNodeStats.put(bucketSpace, bucketSpaceStats);
+        return this;
+    }
+
+    public ContentClusterStatsBuilder add(int nodeIndex) {
+        stats.put(nodeIndex, new HashMap<>());
+        return this;
+    }
+
+    public ContentClusterStats build() {
+        Map<Integer, ContentNodeStats> nodeToStatsMap = new HashMap<>();
+        stats.forEach((nodeIndex, bucketSpaces) ->
+                nodeToStatsMap.put(nodeIndex, new ContentNodeStats(nodeIndex, bucketSpaces)));
+        return new ContentClusterStats(nodeToStatsMap);
+    }
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStatsTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStatsTest.java
new file mode 100644
index 00000000000..c142913a061
--- /dev/null
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/ContentNodeStatsTest.java
@@ -0,0 +1,69 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.clustercontroller.core;
+
+import org.junit.Test;
+
+import static com.yahoo.vespa.clustercontroller.core.ContentNodeStats.BucketSpaceStats;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class ContentNodeStatsTest {
+
+    @Test
+    public void bucket_space_stats_can_transition_from_valid_to_invalid() {
+        BucketSpaceStats stats = BucketSpaceStats.of(5,1);
+        assertTrue(stats.valid());
+
+        stats.merge(BucketSpaceStats.invalid(), 1);
+        assertFalse(stats.valid());
+        assertEquals(BucketSpaceStats.invalid(5, 1), stats);
+    }
+
+    @Test
+    public void bucket_space_stats_can_transition_from_invalid_to_valid() {
+        BucketSpaceStats stats = BucketSpaceStats.invalid();
+        assertFalse(stats.valid());
+
+        stats.merge(BucketSpaceStats.of(5, 1), 1);
+        assertFalse(stats.valid());
+        stats.merge(BucketSpaceStats.invalid(), -1);
+        assertTrue(stats.valid());
+        assertEquals(BucketSpaceStats.of(5, 1), stats);
+    }
+
+    @Test
+    public void bucket_space_stats_tracks_multiple_layers_of_invalid() {
+        BucketSpaceStats stats = BucketSpaceStats.invalid();
+        stats.merge(BucketSpaceStats.invalid(), 1);
+        assertFalse(stats.valid());
+        stats.merge(BucketSpaceStats.invalid(), 1);
+        assertFalse(stats.valid());
+        stats.merge(BucketSpaceStats.of(5, 1), 1);
+        assertFalse(stats.valid());
+
+        stats.merge(BucketSpaceStats.invalid(), -1);
+        assertFalse(stats.valid());
+        stats.merge(BucketSpaceStats.invalid(), -1);
+        assertFalse(stats.valid());
+        stats.merge(BucketSpaceStats.invalid(), -1);
+        assertTrue(stats.valid());
+        assertEquals(BucketSpaceStats.of(5, 1), stats);
+    }
+
+    @Test
+    public void invalid_bucket_space_stats_may_have_pending_buckets() {
+        assertTrue(BucketSpaceStats.invalid().mayHaveBucketsPending());
+    }
+
+    @Test
+    public void valid_bucket_space_stats_may_have_pending_buckets() {
+        assertTrue(BucketSpaceStats.of(5, 1).mayHaveBucketsPending());
+    }
+
+    @Test
+    public void valid_bucket_space_stats_may_have_no_pending_buckets() {
+        assertFalse(BucketSpaceStats.of(5, 0).mayHaveBucketsPending());
+    }
+
+}
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MaintenancehenPendingGlobalMergesTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MaintenancehenPendingGlobalMergesTest.java
index 6c51f251096..fa92a4d5246 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MaintenancehenPendingGlobalMergesTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/MaintenancehenPendingGlobalMergesTest.java
@@ -30,7 +30,7 @@ public class MaintenancehenPendingGlobalMergesTest {
     @Test
     public void no_nodes_set_to_maintenance_in_global_bucket_space_state() {
         Fixture f = new Fixture();
-        when(f.mockPendingChecker.hasMergesPending(eq(globalSpace()), anyInt())).thenReturn(true); // False returned by default otherwise
+        when(f.mockPendingChecker.mayHaveMergesPending(eq(globalSpace()), anyInt())).thenReturn(true); // False returned by default otherwise
         ClusterState derived = f.deriver.derivedFrom(ClusterState.stateFromString("distributor:2 storage:2"), globalSpace());
         assertThat(derived, equalTo(ClusterState.stateFromString("distributor:2 storage:2")));
     }
@@ -38,8 +38,8 @@ public class MaintenancehenPendingGlobalMergesTest {
     @Test
     public void content_nodes_with_global_merge_pending_set_to_maintenance_in_default_space_state() {
         Fixture f = new Fixture();
-        when(f.mockPendingChecker.hasMergesPending(globalSpace(), 1)).thenReturn(true);
-        when(f.mockPendingChecker.hasMergesPending(globalSpace(), 3)).thenReturn(true);
+        when(f.mockPendingChecker.mayHaveMergesPending(globalSpace(), 1)).thenReturn(true);
+        when(f.mockPendingChecker.mayHaveMergesPending(globalSpace(), 3)).thenReturn(true);
         ClusterState derived = f.deriver.derivedFrom(ClusterState.stateFromString("distributor:5 storage:5"), defaultSpace());
         assertThat(derived, equalTo(ClusterState.stateFromString("distributor:5 storage:5 .1.s:m .3.s:m")));
     }
@@ -54,7 +54,7 @@ public class MaintenancehenPendingGlobalMergesTest {
     @Test
     public void default_space_merges_do_not_count_towards_maintenance() {
         Fixture f = new Fixture();
-        when(f.mockPendingChecker.hasMergesPending(eq(defaultSpace()), anyInt())).thenReturn(true);
+        when(f.mockPendingChecker.mayHaveMergesPending(eq(defaultSpace()), anyInt())).thenReturn(true);
         ClusterState derived = f.deriver.derivedFrom(ClusterState.stateFromString("distributor:2 storage:2"), defaultSpace());
         assertThat(derived, equalTo(ClusterState.stateFromString("distributor:2 storage:2")));
     }
@@ -62,7 +62,7 @@ public class MaintenancehenPendingGlobalMergesTest {
     @Test
     public void nodes_only_set_to_maintenance_when_marked_up_init_or_retiring() {
         Fixture f = new Fixture();
-        when(f.mockPendingChecker.hasMergesPending(eq(globalSpace()), anyInt())).thenReturn(true);
+        when(f.mockPendingChecker.mayHaveMergesPending(eq(globalSpace()), anyInt())).thenReturn(true);
         ClusterState derived = f.deriver.derivedFrom(ClusterState.stateFromString("distributor:5 storage:5 .1.s:m .2.s:r .3.s:i .4.s:d"), defaultSpace());
         // TODO reconsider role of retired here... It should not have merges pending towards it in the general case, but may be out of sync
         assertThat(derived, equalTo(ClusterState.stateFromString("distributor:5 storage:5 .0.s:m .1.s:m .2.s:m .3.s:m .4.s:d")));
diff --git a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
index 0f4d1fcdefc..0859ee0e409 100644
--- a/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
+++ b/clustercontroller-core/src/test/java/com/yahoo/vespa/clustercontroller/core/StateVersionTrackerTest.java
@@ -6,6 +6,7 @@ import com.yahoo.vdslib.state.Node;
 import com.yahoo.vdslib.state.NodeState;
 import com.yahoo.vdslib.state.NodeType;
 import com.yahoo.vdslib.state.State;
+import com.yahoo.vespa.clustercontroller.core.hostinfo.HostInfo;
 import org.junit.Test;
 
 import java.text.ParseException;
@@ -17,6 +18,8 @@ import static org.hamcrest.core.Is.is;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class StateVersionTrackerTest {
 
@@ -267,4 +270,52 @@ public class StateVersionTrackerTest {
         assertTrue(versionTracker.candidateChangedEnoughFromCurrentToWarrantPublish());
     }
 
+    @Test
+    public void buckets_pending_state_is_tracked_between_cluster_states() {
+        final StateVersionTracker tracker = createWithMockedMetrics();
+        final NodeInfo distributorNode = mock(DistributorNodeInfo.class);
+        when(distributorNode.isDistributor()).thenReturn(true);
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        tracker.updateLatestCandidateStateBundle(ClusterStateBundle
+                .ofBaselineOnly(stateWithoutAnnotations("distributor:1 storage:1")));
+        tracker.promoteCandidateToVersionedState(1234);
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        // Give 'global' bucket space no buckets pending, which is the same as previous stats
+        tracker.handleUpdatedHostInfo(distributorNode, createHostInfo(0));
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        // Give 'global' bucket space buckets pending, which is different from previous stats
+        tracker.handleUpdatedHostInfo(distributorNode, createHostInfo(1));
+        assertTrue(tracker.bucketSpaceMergeCompletionStateHasChanged());
+
+        tracker.updateLatestCandidateStateBundle(ClusterStateBundle
+                .ofBaselineOnly(stateWithoutAnnotations("distributor:1 storage:1")));
+        assertFalse(tracker.bucketSpaceMergeCompletionStateHasChanged());
+    }
+
+    private HostInfo createHostInfo(long bucketsPending) {
+        return HostInfo.createHostInfo(
+                "{\n" +
+                "\"cluster-state-version\": 2,\n" +
+                "\"distributor\": {\n" +
+                "  \"storage-nodes\": [\n" +
+                "    {\n" +
+                "      \"node-index\": 0,\n" +
+                "      \"bucket-spaces\": [\n" +
+                "        {\n" +
+                "          \"name\": \"global\"\n," +
+                "          \"buckets\": {\n" +
+                "            \"total\": 5,\n" +
+                "            \"pending\": " + bucketsPending + "\n" +
+                "          }\n" +
+                "        }\n" +
+                "      ]\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}\n" +
+                "}");
+    }
+
 }