NodeManager implements INodeManager { @Override public synchronized void addNode(String nodeId, NodeControllerState ncState) throws HyracksException { LOGGER.warn("+addNode: " + nodeId); if (nodeId == null || ncState == null) { throw HyracksException.create(ErrorCode.INVALID_INPUT_PARAMETER); } if (!gatekeeper.isAuthorized(nodeId)) { throw HyracksException.create(ErrorCode.NO_SUCH_NODE, nodeId); } if (nodeRegistry.containsKey(nodeId)) { LOGGER.warn("Node '" + nodeId + "' is already registered; failing the node then re-registering."); failNode(nodeId); } try { ncState.getNodeController().abortJobs(ccs.getCcId()); } catch (IPCException e) { throw HyracksDataException.create(e); } LOGGER.info("adding node to registry"); nodeRegistry.put(nodeId, ncState); try { InetAddress ipAddress = getIpAddress(ncState); Set<String> nodes = ipAddressNodeNameMap.computeIfAbsent(ipAddress, k -> new HashSet<>()); nodes.add(nodeId); } catch (HyracksException e) { nodeRegistry.remove(nodeId); throw e; } LOGGER.info("updating cluster capacity"); resourceManager.update(nodeId, getAdjustedNodeCapacity(ncState.getCapacity())); } NodeManager(ClusterControllerService ccs, CCConfig ccConfig, IResourceManager resourceManager,
            IGatekeeper gatekeeper); @Override Map<InetAddress, Set<String>> getIpAddressNodeNameMap(); @Override Collection<String> getAllNodeIds(); @Override Collection<NodeControllerState> getAllNodeControllerStates(); @Override NodeControllerState getNodeControllerState(String nodeId); @Override synchronized void addNode(String nodeId, NodeControllerState ncState); @Override @Idempotent synchronized void removeNode(String nodeId); @Override Map<String, NodeControllerInfo> getNodeControllerInfoMap(); @Override synchronized Pair<Collection<String>, Collection<JobId>> removeDeadNodes(); synchronized void failNode(String nodeId); @Override void apply(NodeFunction nodeFunction);  }
@Test public void testException() throws HyracksException, IPCException { IResourceManager resourceManager = new ResourceManager(); INodeManager nodeManager = new NodeManager(mockCcs(), makeCCConfig(), resourceManager, nodeId -> true); NodeControllerState ncState1 = mockNodeControllerState(NODE1, true); boolean invalidNetworkAddress = false; try { nodeManager.addNode(NODE1, ncState1); } catch (HyracksException e) { invalidNetworkAddress = e.getErrorCode() == ErrorCode.INVALID_NETWORK_ADDRESS; } Assert.assertTrue(invalidNetworkAddress); verifyEmptyCluster(resourceManager, nodeManager); }
@Test public void testNullNode() throws HyracksException { IResourceManager resourceManager = new ResourceManager(); INodeManager nodeManager = new NodeManager(null, makeCCConfig(), resourceManager, nodeId -> true); boolean invalidParameter = false; try { nodeManager.addNode(null, null); } catch (HyracksException e) { invalidParameter = e.getErrorCode() == ErrorCode.INVALID_INPUT_PARAMETER; } Assert.assertTrue(invalidParameter); verifyEmptyCluster(resourceManager, nodeManager); }