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

        f.ackStateBundleFromBothDistributors();

        final var d0ActivateWaiter = ArgumentCaptor.forClass(Communicator.Waiter.class);
        final var d1ActivateWaiter = ArgumentCaptor.forClass(Communicator.Waiter.class);

        clusterNodeInfos(cf.cluster(), Node.ofDistributor(0), Node.ofDistributor(1)).forEach(nodeInfo -> {
            verify(f.mockCommunicator).activateClusterStateVersion(eq(123), eq(nodeInfo),
                    (nodeInfo.getNodeIndex() == 0 ? d0ActivateWaiter : d1ActivateWaiter).capture());
        });

        respondToActivateClusterStateVersion(cf.cluster.getNodeInfo(Node.ofDistributor(0)),
                f.stateBundle, d0ActivateWaiter.getValue());
        f.simulateBroadcastTick(cf, 123);

        assertNull(f.broadcaster.getLastClusterStateBundleConverged()); // Not yet converged

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

        // Finally, all distributors have ACKed the version! State is marked as converged.
        assertEquals(f.stateBundle, f.broadcaster.getLastClusterStateBundleConverged());
    }