RPCCommunicator implements Communicator { @Override public void setSystemState(ClusterStateBundle stateBundle, NodeInfo node, Waiter<SetClusterStateRequest> externalWaiter) { RPCSetClusterStateWaiter waiter = new RPCSetClusterStateWaiter(externalWaiter, timer); 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; } int nodeVersion = node.getVersion(); Request req; if (nodeVersion <= 2) { req = new Request(LEGACY_SET_SYSTEM_STATE2_RPC_METHOD_NAME); req.parameters().add(new StringValue(baselineState.toString(false))); } else { 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, 60, waiter); node.setClusterStateVersionBundleSent(stateBundle); } 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 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(); assertThat(req, notNullValue()); assertThat(req.methodName(), equalTo(RPCCommunicator.SET_DISTRIBUTION_STATES_RPC_METHOD_NAME)); assertTrue(req.parameters().satisfies("bix")); ClusterStateBundle receivedBundle = RPCUtil.decodeStateBundleFromSetDistributionStatesRequest(req); assertThat(receivedBundle, equalTo(sentBundle)); }
@Test public void set_distribution_states_v3_rpc_auto_downgrades_to_v2_on_unknown_method_error() { var f = new Fixture<SetClusterStateRequest>(); var cf = ClusterFixture.forFlatCluster(3).bringEntireClusterUp().assignDummyRpcAddresses(); var sentBundle = ClusterStateBundleUtil.makeBundle("version:123 distributor:3 storage:3"); f.communicator.setSystemState(sentBundle, cf.cluster().getNodeInfo(Node.ofStorage(1)), f.mockWaiter); RequestWaiter waiter = f.receivedWaiter.get(); assertThat(waiter, notNullValue()); Request req = f.receivedRequest.get(); assertThat(req, notNullValue()); req.setError(ErrorCode.NO_SUCH_METHOD, "que?"); waiter.handleRequestDone(req); cf.cluster().getNodeInfo(Node.ofStorage(1)).setClusterStateBundleVersionAcknowledged(123, false); f.receivedRequest.set(null); f.communicator.setSystemState(sentBundle, cf.cluster().getNodeInfo(Node.ofStorage(1)), f.mockWaiter); req = f.receivedRequest.get(); assertThat(req, notNullValue()); assertThat(req.methodName(), equalTo(RPCCommunicator.LEGACY_SET_SYSTEM_STATE2_RPC_METHOD_NAME)); }