public ClusterStateBundle getLastClusterStateBundleConverged() {
        return lastClusterStateBundleConverged;
    }
@Test
    @SuppressWarnings("unchecked")
    void activation_not_sent_before_all_distributors_have_acked_state_bundle() {
        var f = StateActivationFixture.withTwoPhaseEnabled();
        var cf = f.cf;

        f.expectSetSystemStateInvocationsToBothDistributors();
        f.simulateBroadcastTick(cf, 123);

        // Respond from distributor 0, but not yet from distributor 1
        respondToSetClusterStateBundle(cf.cluster.getNodeInfo(Node.ofDistributor(0)), f.stateBundle, f.d0Waiter.getValue());
        f.simulateBroadcastTick(cf, 123);

        // No activations should be sent yet
        cf.cluster().getNodeInfos().forEach(nodeInfo -> {
            verify(f.mockCommunicator, times(0)).activateClusterStateVersion(eq(123), eq(nodeInfo), any());
        });
        assertNull(f.broadcaster.getLastClusterStateBundleConverged());

        respondToSetClusterStateBundle(cf.cluster.getNodeInfo(Node.ofDistributor(1)), f.stateBundle, f.d1Waiter.getValue());
        f.simulateBroadcastTick(cf, 123);

        // Activation should now be sent to _all_ nodes (distributor and storage)
        cf.cluster().getNodeInfos().forEach(nodeInfo -> {
            verify(f.mockCommunicator).activateClusterStateVersion(eq(123), eq(nodeInfo), any());
        });
        // But not converged yet, as activations have not been ACKed
        assertNull(f.broadcaster.getLastClusterStateBundleConverged());
    }