@Override
    public void setSystemState(ClusterStateBundle stateBundle, NodeInfo node, Waiter<SetClusterStateRequest> externalWaiter) {
        RPCSetClusterStateWaiter waiter = new RPCSetClusterStateWaiter(externalWaiter);
        ClusterState baselineState = stateBundle.getBaselineClusterState();

        Target connection = getConnection(node);
        if ( ! connection.isValid()) {
            log.log(Level.FINE, () -> String.format("Connection to '%s' could not be created.", node.getRpcAddress()));
            return;
        }
        Request req = new Request(SET_DISTRIBUTION_STATES_RPC_METHOD_NAME);
        SlimeClusterStateBundleCodec codec = new SlimeClusterStateBundleCodec();
        EncodedClusterStateBundle encodedBundle = codec.encode(stateBundle);
        Values v = req.parameters();
        v.add(new Int8Value(encodedBundle.getCompression().type().getCode()));
        v.add(new Int32Value(encodedBundle.getCompression().uncompressedSize()));
        v.add(new DataValue(encodedBundle.getCompression().data()));

        log.log(Level.FINE, () -> String.format("Sending '%s' RPC to %s for state version %d",
                req.methodName(), node.getRpcAddress(), stateBundle.getVersion()));
        RPCSetClusterStateRequest stateRequest = new RPCSetClusterStateRequest(node, req, baselineState.getVersion());
        waiter.setRequest(stateRequest);

        connection.invokeAsync(req, Duration.ofSeconds(60), waiter);
        node.setClusterStateVersionBundleSent(stateBundle);
    }
@Test
    void setSystemState_v3_sends_distribution_states_rpc() {
        var f = new Fixture<SetClusterStateRequest>();
        var cf = ClusterFixture.forFlatCluster(3).bringEntireClusterUp().assignDummyRpcAddresses();
        var sentBundle = ClusterStateBundleUtil.makeBundle("distributor:3 storage:3");
        f.communicator.setSystemState(sentBundle, cf.cluster().getNodeInfo(Node.ofStorage(1)), f.mockWaiter);

        Request req = f.receivedRequest.get();
        assertNotNull(req);
        assertEquals(req.methodName(), RPCCommunicator.SET_DISTRIBUTION_STATES_RPC_METHOD_NAME);
        assertTrue(req.parameters().satisfies("bix")); // <compression type>, <uncompressed size>, <payload>

        ClusterStateBundle receivedBundle = RPCUtil.decodeStateBundleFromSetDistributionStatesRequest(req);
        assertEquals(receivedBundle, sentBundle);
    }