void checkIfClusterStateIsAckedByAllDistributors(DatabaseHandler database,
                                                     DatabaseHandler.DatabaseContext dbContext,
                                                     FleetController fleetController) throws InterruptedException {
        if ((clusterStateBundle == null) || currentClusterStateIsConverged()) {
            return; // Nothing to do for the current state
        }
        final int currentStateVersion = clusterStateBundle.getVersion();
        boolean anyDistributorsNeedStateBundle = dbContext.getCluster().getNodeInfos().stream()
                                                          .filter(NodeInfo::isDistributor)
                                                          .anyMatch(this::nodeNeedsClusterStateBundle);

        if (!anyDistributorsNeedStateBundle && (currentStateVersion > lastStateVersionBundleAcked)) {
            markCurrentClusterStateBundleAsReceivedByAllDistributors();
            if (clusterStateBundle.deferredActivation()) {
                context.log(log,
                            Level.FINE,
                            () -> String.format("All distributors have ACKed cluster state " +
                                                "version %d, sending activation", currentStateVersion));
            } else {
                markCurrentClusterStateAsConverged(database, dbContext, fleetController);
            }
            return; // Either converged (no two-phase) or activations must be sent before we can continue.
        }

        if (anyDistributorsNeedStateBundle || !clusterStateBundle.deferredActivation()) {
            return;
        }

        boolean anyDistributorsNeedActivation = dbContext.getCluster().getNodeInfos().stream()
                                                         .filter(NodeInfo::isDistributor)
                                                         .anyMatch(this::nodeNeedsClusterStateActivation);

        if (!anyDistributorsNeedActivation && (currentStateVersion > lastClusterStateVersionConverged)) {
            markCurrentClusterStateAsConverged(database, dbContext, fleetController);
        } else {
            context.log(log,
                        Level.FINE,
                        () -> String.format("distributors still need activation in state %d (last converged: %d)",
                                            currentStateVersion, lastClusterStateVersionConverged));
        }
    }
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