RPCCommunicator implements Communicator { protected int generateNodeStateRequestTimeoutMs() { double intervalFraction = Math.random(); double earliestTimeoutSeconds = nodeStateRequestTimeoutIntervalMaxSeconds * nodeStateRequestTimeoutIntervalStartPercentage / 100.0; double latestTimeoutSeconds = nodeStateRequestTimeoutIntervalMaxSeconds * nodeStateRequestTimeoutIntervalStopPercentage / 100.0; double interval = latestTimeoutSeconds - earliestTimeoutSeconds; double timeoutSeconds = earliestTimeoutSeconds + intervalFraction * interval; return (int) (timeoutSeconds * 1000); } 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 testGenerateNodeStateRequestTimeoutMs() { final RPCCommunicator communicator = new RPCCommunicator( RPCCommunicator.createRealSupervisor(), null , INDEX, NODE_STATE_REQUEST_TIMEOUT_INTERVAL_MAX_MS, NODE_STATE_REQUEST_TIMEOUT_INTERVAL_START_PERCENTAGE, NODE_STATE_REQUEST_TIMEOUT_INTERVAL_STOP_PERCENTAGE, 0); int max = -1; int min = 100000; final Set<Integer> uniqueTimeoutValues = new HashSet<>(); for (int x = 0; x < TEST_ITERATIONS; x++) { int timeOutMs = communicator.generateNodeStateRequestTimeoutMs(); min = Math.min(min, timeOutMs); max = Math.max(max, timeOutMs); uniqueTimeoutValues.add(timeOutMs); } assertTrue(max <= NODE_STATE_REQUEST_TIMEOUT_INTERVAL_MAX_MS * NODE_STATE_REQUEST_TIMEOUT_INTERVAL_STOP_PERCENTAGE / 100.); assertThat(min, is(not(max))); assertTrue(min >= NODE_STATE_REQUEST_TIMEOUT_INTERVAL_START_PERCENTAGE * NODE_STATE_REQUEST_TIMEOUT_INTERVAL_MAX_MS / 100); assertTrue(uniqueTimeoutValues.size()> TEST_ITERATIONS/2); }