public boolean broadcastNewStateBundleIfRequired(DatabaseHandler.DatabaseContext dbContext, Communicator communicator,
                                                     int lastClusterStateVersionWrittenToZooKeeper) {
        if (clusterStateBundle == null || clusterStateBundle.getVersion() == 0) {
            return false;
        }
        if (clusterStateBundle.getVersion() != lastClusterStateVersionWrittenToZooKeeper) {
            return false;
        }

        int baselineStateVersion = clusterStateBundle.getBaselineClusterState().getVersion();

        if (!currentBundleVersionIsTaggedOfficial()) {
            context.log(log, Level.INFO, "Publishing cluster state version " + baselineStateVersion);
            tagCurrentBundleVersionAsOfficial();
        }

        List<NodeInfo> recipients = resolveStateVersionSendSet(dbContext);
        ClusterStateBundle modifiedBundle = clusterStateBundle.cloneWithMapper(state -> buildModifiedClusterState(state, dbContext));
        for (NodeInfo node : recipients) {
            if (nodeNeedsToObserveStartupTimestamps(node)) {
                context.log(log,
                            Level.FINE,
                            () -> "Sending modified cluster state version " + baselineStateVersion +
                                  " to node " + node + ": " + modifiedBundle);
                communicator.setSystemState(modifiedBundle, node, setClusterStateWaiter);
            } else {
                context.log(log,
                            Level.FINE,
                            () -> "Sending system state version " + baselineStateVersion +
                                  " to node " + node + ". (went down time " + node.getWentDownWithStartTime() +
                                  ", node start time " + node.getStartTimestamp() + ")");
                communicator.setSystemState(clusterStateBundle, node, setClusterStateWaiter);
            }
        }

        return !recipients.isEmpty();
    }
Checks if all distributor nodes have ACKed (and activated) the most recent cluster state.
Iff this is the case, triggers handleAllDistributorsInSync() on the provided FleetController
object and updates the broadcaster's last known in-sync cluster state version.
clustercontroller-core/src/main/java/com/yahoo/vespa/clustercontroller/core/SystemStateBroadcaster.java