@Test(timeout = 5000) public void testRequestNodeDisconnect() throws InterruptedException { final NodeIdentifier nodeId1 = createNodeId(1); coordinator.updateNodeStatus(new NodeConnectionStatus(nodeId1, NodeConnectionState.CONNECTED)); coordinator.updateNodeStatus(new NodeConnectionStatus(createNodeId(2), NodeConnectionState.CONNECTED)); while (nodeStatuses.isEmpty()) { Thread.sleep(10L); } nodeStatuses.clear(); coordinator.requestNodeDisconnect(nodeId1, DisconnectionCode.USER_DISCONNECTED, "Unit Test"); assertEquals(NodeConnectionState.DISCONNECTED, coordinator.getConnectionStatus(nodeId1).getState()); while (nodeStatuses.isEmpty()) { Thread.sleep(10L); } final NodeConnectionStatus status = nodeStatuses.get(0); assertEquals(nodeId1, status.getNodeIdentifier()); assertEquals(NodeConnectionState.DISCONNECTED, status.getState()); }
@Override public void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) { throw new IllegalNodeDisconnectionException("Cannot disconnect node " + nodeId + " because it is the only node currently connected"); } logger.info("Requesting that {} disconnect due to {}", nodeId, explanation == null ? disconnectionCode : explanation); updateNodeStatus(new NodeConnectionStatus(nodeId, disconnectionCode, explanation)); if (disconnectionCode == DisconnectionCode.NODE_SHUTDOWN) { return; } final DisconnectMessage request = new DisconnectMessage(); request.setNodeId(nodeId); request.setExplanation(explanation); addNodeEvent(nodeId, "Disconnection requested due to " + explanation); disconnectAsynchronously(request, 10, 5); }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) { throw new IllegalNodeDisconnectionException("Cannot disconnect node " + nodeId + " because it is the only node currently connected"); } logger.info("Requesting that {} disconnect due to {}", nodeId, explanation == null ? disconnectionCode : explanation); updateNodeStatus(new NodeConnectionStatus(nodeId, disconnectionCode, explanation)); if (disconnectionCode == DisconnectionCode.NODE_SHUTDOWN) { return; } final DisconnectMessage request = new DisconnectMessage(); request.setNodeId(nodeId); request.setExplanation(explanation); addNodeEvent(nodeId, "Disconnection requested due to " + explanation); disconnectAsynchronously(request, 10, 5); } }
NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { @Override public void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) { throw new IllegalNodeDisconnectionException("Cannot disconnect node " + nodeId + " because it is the only node currently connected"); } logger.info("Requesting that {} disconnect due to {}", nodeId, explanation == null ? disconnectionCode : explanation); updateNodeStatus(new NodeConnectionStatus(nodeId, disconnectionCode, explanation)); if (disconnectionCode == DisconnectionCode.NODE_SHUTDOWN) { return; } final DisconnectMessage request = new DisconnectMessage(); request.setNodeId(nodeId); request.setExplanation(explanation); addNodeEvent(nodeId, "Disconnection requested due to " + explanation); disconnectAsynchronously(request, 10, 5); } 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 void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) { throw new IllegalNodeDisconnectionException("Cannot disconnect node " + nodeId + " because it is the only node currently connected"); } logger.info("Requesting that {} disconnect due to {}", nodeId, explanation == null ? disconnectionCode : explanation); updateNodeStatus(new NodeConnectionStatus(nodeId, disconnectionCode, explanation)); if (disconnectionCode == DisconnectionCode.NODE_SHUTDOWN) { return; } final DisconnectMessage request = new DisconnectMessage(); request.setNodeId(nodeId); request.setExplanation(explanation); addNodeEvent(nodeId, "Disconnection requested due to " + explanation); disconnectAsynchronously(request, 10, 5); } 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 void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) { throw new IllegalNodeDisconnectionException("Cannot disconnect node " + nodeId + " because it is the only node currently connected"); } logger.info("Requesting that {} disconnect due to {}", nodeId, explanation == null ? disconnectionCode : explanation); updateNodeStatus(new NodeConnectionStatus(nodeId, disconnectionCode, explanation)); if (disconnectionCode == DisconnectionCode.NODE_SHUTDOWN) { return; } final DisconnectMessage request = new DisconnectMessage(); request.setNodeId(nodeId); request.setExplanation(explanation); addNodeEvent(nodeId, "Disconnection requested due to " + explanation); disconnectAsynchronously(request, 10, 5); } 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();  }