### Question:
ClusterStateAggregator { public ClusterState getAggregatedState() { if (clusterStates.isEmpty()) { throw new IllegalStateException("Insufficient cluster state"); } if (clusterStates.size() == 1) { return clusterStates.get(0); } Map<String, NodeState> stateMap = clusterStates.stream() .map(ClusterState::getNodes) .reduce((prevNodes, currNodes) -> { Map<String, NodeState> actualState = new HashMap<>(); for (String endpoint : currNodes.keySet()) { NodeState prevNodeState = prevNodes.get(endpoint); NodeState currNodeState = currNodes.get(endpoint); if (currNodeState.isConnected() || !prevNodeState.isConnected()) { actualState.put(endpoint, currNodeState); } else { actualState.put(endpoint, prevNodeState); } } return ImmutableMap.copyOf(actualState); }).orElse(ImmutableMap.of()); return ClusterState.builder() .localEndpoint(localEndpoint) .unresponsiveNodes(unresponsiveNodes) .nodes(stateMap) .build(); }  ClusterState getAggregatedState();  }### Answer:
@Test public void getAggregatedStateSingleNodeCluster() { final String localEndpoint = "a"; ClusterState clusterState = ClusterState.buildClusterState( localEndpoint, ImmutableList.of(), nodeState("a", epoch, OK) ); List<ClusterState> clusterStates = Arrays.asList(clusterState, clusterState, clusterState); ClusterStateAggregator aggregator = ClusterStateAggregator.builder() .localEndpoint(localEndpoint) .clusterStates(clusterStates) .unresponsiveNodes(ImmutableList.of()) .build(); NodeState expectedNodeState = clusterState.getNode("a").get(); NodeState actualNodeState = aggregator.getAggregatedState().getNode("a").get(); assertThat(actualNodeState).isEqualTo(expectedNodeState); }
@Test public void getAggregatedState() { final String localEndpoint = "a"; ClusterState clusterState1 = ClusterState.buildClusterState( localEndpoint, ImmutableList.of(), nodeState("a", epoch, OK, FAILED, FAILED), NodeState.getUnavailableNodeState("b"), NodeState.getUnavailableNodeState("c") ); ClusterState clusterState2 = ClusterState.buildClusterState( localEndpoint, ImmutableList.of(), nodeState("a", epoch, OK, OK, FAILED), nodeState("b", epoch, OK, OK, FAILED), NodeState.getUnavailableNodeState("c") ); final int epoch = 1; final int counter = 123; ClusterState clusterState3 = ClusterState.buildClusterState( localEndpoint, ImmutableList.of(), nodeState("a", epoch, OK, FAILED, FAILED), NodeState.getUnavailableNodeState("b"), NodeState.getNotReadyNodeState("c") ); List<ClusterState> clusterStates = Arrays.asList( clusterState1, clusterState2, clusterState3 ); ClusterStateAggregator aggregator = ClusterStateAggregator.builder() .localEndpoint(localEndpoint) .clusterStates(clusterStates) .unresponsiveNodes(ImmutableList.of()) .build(); NodeState expectedLocalNodeState = clusterState3.getNode(localEndpoint).get(); NodeState localNodeState = aggregator.getAggregatedState().getNode(localEndpoint).get(); assertThat(localNodeState.isConnected()).isTrue(); assertThat(localNodeState).isEqualTo(expectedLocalNodeState); NodeState expectedNodeBState = clusterState2.getNode("b").get(); NodeState nodeBState = aggregator.getAggregatedState().getNode("b").get(); assertThat(nodeBState.isConnected()).isTrue(); assertThat(nodeBState).isEqualTo(expectedNodeBState); NodeState expectedNodeCState = clusterState3.getNode("c").get(); NodeState nodeCState = aggregator.getAggregatedState().getNode("c").get(); assertThat(nodeCState.isConnected()).isFalse(); assertThat(nodeCState).isEqualTo(expectedNodeCState); }