DiscoveryService implements Runnable { public static InetAddress getTaskManagerAddress(final InetAddress jobManagerAddress) throws DiscoveryException { final int magicNumber = GlobalConfiguration.getInteger(MAGICNUMBER_KEY, DEFAULT_MAGICNUMBER); final int discoveryPort = GlobalConfiguration.getInteger(DISCOVERYPORT_KEY, DEFAULT_DISCOVERYPORT); InetAddress taskManagerAddress = null; DatagramSocket socket = null; try { socket = new DatagramSocket(); LOG.debug("Setting socket timeout to " + CLIENTSOCKETTIMEOUT); socket.setSoTimeout(CLIENTSOCKETTIMEOUT); final DatagramPacket responsePacket = new DatagramPacket(new byte[RESPONSE_PACKET_SIZE], RESPONSE_PACKET_SIZE); for (int retries = 0; retries < DISCOVERFAILURERETRIES; retries++) { final DatagramPacket addressRequest = createTaskManagerAddressRequestPacket(magicNumber); addressRequest.setAddress(jobManagerAddress); addressRequest.setPort(discoveryPort); LOG.debug("Sending Task Manager address request to " + addressRequest.getSocketAddress()); socket.send(addressRequest); try { socket.receive(responsePacket); } catch (SocketTimeoutException ste) { LOG.warn("Timeout wainting for task manager address reply. Retrying..."); continue; } if (!isPacketForUs(responsePacket, magicNumber)) { LOG.warn("Received packet which is not destined to this Nephele setup"); continue; } final int packetTypeID = getPacketTypeID(responsePacket); if (packetTypeID != TM_ADDRESS_REPLY_ID) { LOG.warn("Received response of unknown type " + packetTypeID + ", discarding..."); continue; } taskManagerAddress = extractInetAddress(responsePacket); break; } } catch (IOException ioe) { throw new DiscoveryException(StringUtils.stringifyException(ioe)); } finally { if (socket != null) { socket.close(); } } if (taskManagerAddress == null) { throw new DiscoveryException("Unable to obtain task manager address"); } return taskManagerAddress; } private  DiscoveryService(final InetAddress ipcAddress, final int ipcPort); static synchronized void startDiscoveryService(final InetAddress ipcAddress, final int ipcPort); boolean isRunning(); static synchronized void stopDiscoveryService(); static InetAddress getTaskManagerAddress(final InetAddress jobManagerAddress); static InetSocketAddress getJobManagerAddress(); @Override void run();  }
@Test public void testTaskManagerAddressResolution() { InetAddress localHost = null; try { localHost = InetAddress.getLocalHost(); } catch(UnknownHostException e) { fail(e.getMessage()); } try { final InetAddress taskManagerAddress = DiscoveryService.getTaskManagerAddress(localHost); assertEquals(localHost, taskManagerAddress); } catch(DiscoveryException e) { fail(e.getMessage()); } }