RPCCommunicator implements Communicator { @Override public void activateClusterStateVersion(int clusterStateVersion, NodeInfo node, Waiter<ActivateClusterStateVersionRequest> externalWaiter) { var waiter = new RPCActivateClusterStateVersionWaiter(externalWaiter); Target connection = getConnection(node); if ( ! connection.isValid()) { log.log(Level.FINE, () -> String.format("Connection to '%s' could not be created.", node.getRpcAddress())); return; } var req = new Request(ACTIVATE_CLUSTER_STATE_VERSION_RPC_METHOD_NAME); req.parameters().add(new Int32Value(clusterStateVersion)); log.log(Level.FINE, () -> String.format("Sending '%s' RPC to %s for state version %d", req.methodName(), node.getRpcAddress(), clusterStateVersion)); var activationRequest = new RPCActivateClusterStateVersionRequest(node, req, clusterStateVersion); waiter.setRequest(activationRequest); connection.invokeAsync(req, 60, waiter); node.setClusterStateVersionActivationSent(clusterStateVersion); } RPCCommunicator(Supervisor supervisor,
                           Timer t,
                           int index,
                           int nodeStateRequestTimeoutIntervalMaxMs,
                           int nodeStateRequestTimeoutIntervalStartPercentage,
                           int nodeStateRequestTimeoutIntervalStopPercentage,
                           int nodeStateRequestRoundTripTimeMaxSeconds); static Supervisor createRealSupervisor(); void shutdown(); Target getConnection(final NodeInfo node); @Override void propagateOptions(FleetControllerOptions options); @Override void getNodeState(NodeInfo node, Waiter<GetNodeStateRequest> externalWaiter); @Override void setSystemState(ClusterStateBundle stateBundle, NodeInfo node, Waiter<SetClusterStateRequest> externalWaiter); @Override void activateClusterStateVersion(int clusterStateVersion, NodeInfo node, Waiter<ActivateClusterStateVersionRequest> externalWaiter); static final Logger log; static final int ACTIVATE_CLUSTER_STATE_VERSION_RPC_VERSION; static final String ACTIVATE_CLUSTER_STATE_VERSION_RPC_METHOD_NAME; static final int SET_DISTRIBUTION_STATES_RPC_VERSION; static final String SET_DISTRIBUTION_STATES_RPC_METHOD_NAME; static final int LEGACY_SET_SYSTEM_STATE2_RPC_VERSION; static final String LEGACY_SET_SYSTEM_STATE2_RPC_METHOD_NAME; }
@Test public void activateClusterStateVersion_sends_version_activation_rpc() { var f = new Fixture<ActivateClusterStateVersionRequest>(); var cf = ClusterFixture.forFlatCluster(3).bringEntireClusterUp().assignDummyRpcAddresses(); f.communicator.activateClusterStateVersion(12345, cf.cluster().getNodeInfo(Node.ofDistributor(1)), f.mockWaiter); Request req = f.receivedRequest.get(); assertThat(req, notNullValue()); assertThat(req.methodName(), equalTo(RPCCommunicator.ACTIVATE_CLUSTER_STATE_VERSION_RPC_METHOD_NAME)); assertTrue(req.parameters().satisfies("i")); assertThat(req.parameters().get(0).asInt32(), equalTo(12345)); }