@Test public void testGetConnectionStates() throws IOException { coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(1), DisconnectionCode.LACK_OF_HEARTBEAT)); coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(2), NodeConnectionState.DISCONNECTING)); coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(3), NodeConnectionState.CONNECTING)); coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(4), NodeConnectionState.CONNECTED)); coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(5), NodeConnectionState.CONNECTED)); final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = coordinator.getConnectionStates(); assertEquals(4, stateMap.size()); final List<NodeIdentifier> connectedIds = stateMap.get(NodeConnectionState.CONNECTED); assertEquals(2, connectedIds.size()); assertTrue(connectedIds.contains(createNodeId(4))); assertTrue(connectedIds.contains(createNodeId(5))); final List<NodeIdentifier> connectingIds = stateMap.get(NodeConnectionState.CONNECTING); assertEquals(1, connectingIds.size()); assertTrue(connectingIds.contains(createNodeId(3))); final List<NodeIdentifier> disconnectingIds = stateMap.get(NodeConnectionState.DISCONNECTING); assertEquals(1, disconnectingIds.size()); assertTrue(disconnectingIds.contains(createNodeId(2))); final List<NodeIdentifier> disconnectedIds = stateMap.get(NodeConnectionState.DISCONNECTED); assertEquals(1, disconnectedIds.size()); assertTrue(disconnectedIds.contains(createNodeId(1))); }
@Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); for (final Map.Entry<NodeIdentifier, NodeConnectionStatus> entry : nodeStatuses.entrySet()) { final NodeConnectionState state = entry.getValue().getState(); final List<NodeIdentifier> nodeIds = connectionStates.computeIfAbsent(state, s -> new ArrayList<NodeIdentifier>()); nodeIds.add(entry.getKey()); } return connectionStates; }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); for (final Map.Entry<NodeIdentifier, NodeConnectionStatus> entry : nodeStatuses.entrySet()) { final NodeConnectionState state = entry.getValue().getState(); final List<NodeIdentifier> nodeIds = connectionStates.computeIfAbsent(state, s -> new ArrayList<NodeIdentifier>()); nodeIds.add(entry.getKey()); } return connectionStates; } }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); for (final Map.Entry<NodeIdentifier, NodeConnectionStatus> entry : nodeStatuses.entrySet()) { final NodeConnectionState state = entry.getValue().getState(); final List<NodeIdentifier> nodeIds = connectionStates.computeIfAbsent(state, s -> new ArrayList<NodeIdentifier>()); nodeIds.add(entry.getKey()); } return connectionStates; } NodeClusterCoordinator(final ClusterCoordinationProtocolSenderListener senderListener, final EventReporter eventReporter, final LeaderElectionManager leaderElectionManager,
            final FlowElection flowElection, final ClusterNodeFirewall firewall, final RevisionManager revisionManager, final NiFiProperties nifiProperties,
            final NodeProtocolSender nodeProtocolSender); }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); for (final Map.Entry<NodeIdentifier, NodeConnectionStatus> entry : nodeStatuses.entrySet()) { final NodeConnectionState state = entry.getValue().getState(); final List<NodeIdentifier> nodeIds = connectionStates.computeIfAbsent(state, s -> new ArrayList<NodeIdentifier>()); nodeIds.add(entry.getKey()); } return connectionStates; } NodeClusterCoordinator(final ClusterCoordinationProtocolSenderListener senderListener, final EventReporter eventReporter, final LeaderElectionManager leaderElectionManager,
            final FlowElection flowElection, final ClusterNodeFirewall firewall, final RevisionManager revisionManager, final NiFiProperties nifiProperties,
            final NodeProtocolSender nodeProtocolSender); @Override void shutdown(); @Override void setLocalNodeIdentifier(final NodeIdentifier nodeId); @Override NodeIdentifier getLocalNodeIdentifier(); @Override void resetNodeStatuses(final Map<NodeIdentifier, NodeConnectionStatus> statusMap); @Override boolean resetNodeStatus(final NodeConnectionStatus connectionStatus, final long qualifyingUpdateId); @Override void requestNodeConnect(final NodeIdentifier nodeId, final String userDn); @Override void finishNodeConnection(final NodeIdentifier nodeId); @Override void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation); @Override void disconnectionRequestedByNode(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation); @Override void removeNode(final NodeIdentifier nodeId, final String userDn); @Override NodeConnectionStatus getConnectionStatus(final NodeIdentifier nodeId); @Override List<NodeConnectionStatus> getConnectionStatuses(); @Override Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates(); @Override boolean isBlockedByFirewall(final String hostname); @Override void reportEvent(final NodeIdentifier nodeId, final Severity severity, final String event); @Override NodeIdentifier getNodeIdentifier(final String uuid); @Override Set<NodeIdentifier> getNodeIdentifiers(final NodeConnectionState... states); @Override NodeIdentifier getPrimaryNode(); @Override NodeIdentifier getElectedActiveCoordinatorNode(); @Override boolean isActiveClusterCoordinator(); @Override List<NodeEvent> getNodeEvents(final NodeIdentifier nodeId); @Override void setFlowService(final FlowService flowService); @Override ProtocolMessage handle(final ProtocolMessage protocolMessage); @Override String getFlowElectionStatus(); @Override boolean isFlowElectionComplete(); @Override boolean canHandle(final ProtocolMessage msg); @Override void afterRequest(final String uriPath, final String method, final Set<NodeResponse> nodeResponses); @Override void setConnected(final boolean connected); @Override boolean isConnected(); @Override Map<NodeIdentifier, NodeWorkload> getClusterWorkload(); }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); for (final Map.Entry<NodeIdentifier, NodeConnectionStatus> entry : nodeStatuses.entrySet()) { final NodeConnectionState state = entry.getValue().getState(); final List<NodeIdentifier> nodeIds = connectionStates.computeIfAbsent(state, s -> new ArrayList<NodeIdentifier>()); nodeIds.add(entry.getKey()); } return connectionStates; } NodeClusterCoordinator(final ClusterCoordinationProtocolSenderListener senderListener, final EventReporter eventReporter, final LeaderElectionManager leaderElectionManager,
            final FlowElection flowElection, final ClusterNodeFirewall firewall, final RevisionManager revisionManager, final NiFiProperties nifiProperties,
            final NodeProtocolSender nodeProtocolSender); @Override void shutdown(); @Override void setLocalNodeIdentifier(final NodeIdentifier nodeId); @Override NodeIdentifier getLocalNodeIdentifier(); @Override void resetNodeStatuses(final Map<NodeIdentifier, NodeConnectionStatus> statusMap); @Override boolean resetNodeStatus(final NodeConnectionStatus connectionStatus, final long qualifyingUpdateId); @Override void requestNodeConnect(final NodeIdentifier nodeId, final String userDn); @Override void finishNodeConnection(final NodeIdentifier nodeId); @Override void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation); @Override void disconnectionRequestedByNode(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation); @Override void removeNode(final NodeIdentifier nodeId, final String userDn); @Override NodeConnectionStatus getConnectionStatus(final NodeIdentifier nodeId); @Override List<NodeConnectionStatus> getConnectionStatuses(); @Override Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates(); @Override boolean isBlockedByFirewall(final String hostname); @Override void reportEvent(final NodeIdentifier nodeId, final Severity severity, final String event); @Override NodeIdentifier getNodeIdentifier(final String uuid); @Override Set<NodeIdentifier> getNodeIdentifiers(final NodeConnectionState... states); @Override NodeIdentifier getPrimaryNode(); @Override NodeIdentifier getElectedActiveCoordinatorNode(); @Override boolean isActiveClusterCoordinator(); @Override List<NodeEvent> getNodeEvents(final NodeIdentifier nodeId); @Override void setFlowService(final FlowService flowService); @Override ProtocolMessage handle(final ProtocolMessage protocolMessage); @Override String getFlowElectionStatus(); @Override boolean isFlowElectionComplete(); @Override boolean canHandle(final ProtocolMessage msg); @Override void afterRequest(final String uriPath, final String method, final Set<NodeResponse> nodeResponses); @Override void setConnected(final boolean connected); @Override boolean isConnected(); @Override Map<NodeIdentifier, NodeWorkload> getClusterWorkload();  }