SystemStateBroadcaster { public ClusterStateBundle getLastClusterStateBundleConverged() { return lastClusterStateBundleConverged; } SystemStateBroadcaster(Timer timer, Object monitor); void handleNewClusterStates(ClusterStateBundle state); ClusterState getClusterState(); boolean hasBroadcastedClusterStateBundle(); void resetBroadcastedClusterStateBundle(); ClusterStateBundle getClusterStateBundle(); ClusterStateBundle getLastClusterStateBundleConverged(); boolean processResponses(); boolean broadcastNewStateBundleIfRequired(DatabaseHandler.Context dbContext, Communicator communicator); boolean broadcastStateActivationsIfRequired(DatabaseHandler.Context dbContext, Communicator communicator); int lastClusterStateVersionInSync(); static Logger log; }
@Test @SuppressWarnings("unchecked") public void activation_not_sent_before_all_distributors_have_acked_state_bundle() { var f = StateActivationFixture.withTwoPhaseEnabled(); var cf = f.cf; f.expectSetSystemStateInvocationsToBothDistributors(); f.simulateBroadcastTick(cf); respondToSetClusterStateBundle(cf.cluster.getNodeInfo(Node.ofDistributor(0)), f.stateBundle, f.d0Waiter.getValue()); f.simulateBroadcastTick(cf); cf.cluster().getNodeInfo().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); cf.cluster().getNodeInfo().forEach(nodeInfo -> { verify(f.mockCommunicator).activateClusterStateVersion(eq(123), eq(nodeInfo), any()); }); assertNull(f.broadcaster.getLastClusterStateBundleConverged()); }
@Test @SuppressWarnings("unchecked") public 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); assertNull(f.broadcaster.getLastClusterStateBundleConverged()); respondToActivateClusterStateVersion(cf.cluster.getNodeInfo(Node.ofDistributor(1)), f.stateBundle, d1ActivateWaiter.getValue()); f.simulateBroadcastTick(cf); assertEquals(f.stateBundle, f.broadcaster.getLastClusterStateBundleConverged()); }
@Test @SuppressWarnings("unchecked") public void activation_not_sent_if_deferred_activation_is_disabled_in_state_bundle() { var f = StateActivationFixture.withTwoPhaseDisabled(); var cf = f.cf; f.ackStateBundleFromBothDistributors(); assertEquals(f.stateBundle, f.broadcaster.getLastClusterStateBundleConverged()); clusterNodeInfos(cf.cluster(), Node.ofDistributor(0), Node.ofDistributor(1)).forEach(nodeInfo -> { verify(f.mockCommunicator, times(0)).activateClusterStateVersion(eq(123), eq(nodeInfo), any()); }); }
@Test @SuppressWarnings("unchecked") public void activation_convergence_considers_actual_version_returned_from_node() { 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()); respondToActivateClusterStateVersion(cf.cluster.getNodeInfo(Node.ofDistributor(1)), f.stateBundle, 124, d1ActivateWaiter.getValue()); f.simulateBroadcastTick(cf); assertNull(f.broadcaster.getLastClusterStateBundleConverged()); }