@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, Duration.ofSeconds(60), waiter);
        node.setClusterStateVersionActivationSent(clusterStateVersion);
    }
@Test
    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();
        assertNotNull(req);
        assertEquals(req.methodName(), RPCCommunicator.ACTIVATE_CLUSTER_STATE_VERSION_RPC_METHOD_NAME);
        assertTrue(req.parameters().satisfies("i")); // <cluster state version>
        assertEquals(req.parameters().get(0).asInt32(), 12345);
    }