[FLINK-2781] [core] Cleanup NetUtils.

- The NetUtils class (in flink-core) contains all methods usable without runtime dependency - The runtime NetUtils class (to find connectiong addresses) is now called ConnectionUtil.
ishark · Oct 1, 2015 · a3c0b44 · a3c0b44
1 parent 28d36cc
commit a3c0b44
Show file tree

Hide file tree

Showing 22 changed files with 66 additions and 83 deletions.
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
@@ -22,7 +22,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.jobgraph.JobVertex;
 import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.util.NetUtils;
 import org.junit.Test;
 
 import java.net.InetAddress;

diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -22,7 +22,6 @@
 import akka.actor.Props;
 import akka.actor.Status;
 
-
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.JobSubmissionResult;
 import org.apache.flink.api.common.Plan;
@@ -44,20 +43,17 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.jobmanager.JobManager;
 import org.apache.flink.runtime.messages.JobManagerMessages;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.runtime.util.SerializedThrowable;
+import org.apache.flink.util.NetUtils;
 
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
-
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 import java.util.Collections;
-
-
 import java.util.UUID;
 
 import static org.junit.Assert.*;

diff --git a/...k-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java b/...k-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/DataStreamUtils.java
@@ -27,7 +27,7 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.runtime.net.ConnectionUtils;
 
 public final class DataStreamUtils {
 
@@ -46,7 +46,7 @@ public static <OUT> Iterator<OUT> collect(DataStream<OUT> stream) {
 			String host = ((RemoteStreamEnvironment)env).getHost();
 			int port = ((RemoteStreamEnvironment)env).getPort();
 			try {
-				clientAddress = NetUtils.findConnectingAddress(new InetSocketAddress(host, port), 2000, 400);
+				clientAddress = ConnectionUtils.findConnectingAddress(new InetSocketAddress(host, port), 2000, 400);
 			} catch (IOException e) {
 				throw new RuntimeException("IOException while trying to connect to the master", e);
 			}

diff --git a/flink-core/src/main/java/org/apache/flink/util/NetUtils.java b/flink-core/src/main/java/org/apache/flink/util/NetUtils.java
@@ -18,7 +18,9 @@
 package org.apache.flink.util;
 
 
+import java.io.IOException;
 import java.net.MalformedURLException;
+import java.net.ServerSocket;
 import java.net.URL;
 
 public class NetUtils {
@@ -65,4 +67,23 @@ public static URL getCorrectHostnamePort(String hostPort) {
 			throw new IllegalArgumentException("The given host:port ('"+hostPort+"') is invalid", e);
 		}
 	}
+
+	/**
+	 * Find a non-occupied port.
+	 *
+	 * @return A non-occupied port.
+	 */
+	public static int getAvailablePort() {
+		for (int i = 0; i < 50; i++) {
+			try (ServerSocket serverSocket = new ServerSocket(0)) {
+				int port = serverSocket.getLocalPort();
+				if (port != 0) {
+					return port;
+				}
+			}
+			catch (IOException ignored) {}
+		}
+
+		throw new RuntimeException("Could not find a free permitted port on the machine.");
+	}
 }
diff --git a/...rg/apache/flink/runtime/net/NetUtils.java → ...he/flink/runtime/net/ConnectionUtils.java b/...rg/apache/flink/runtime/net/NetUtils.java → ...he/flink/runtime/net/ConnectionUtils.java
@@ -23,7 +23,6 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.NetworkInterface;
-import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketAddress;
 import java.util.Enumeration;
@@ -41,9 +40,9 @@
  * Utilities to determine the network interface and address that should be used to bind the
  * TaskManager communication to.
  */
-public class NetUtils {
+public class ConnectionUtils {
 
-	private static final Logger LOG = LoggerFactory.getLogger(NetUtils.class);
+	private static final Logger LOG = LoggerFactory.getLogger(ConnectionUtils.class);
 
 	private static final long MIN_SLEEP_TIME = 50;
 	private static final long MAX_SLEEP_TIME = 20000;
@@ -282,8 +281,7 @@ private static boolean tryToConnect(InetAddress fromAddress, SocketAddress toSoc
 			LOG.debug("Trying to connect to (" + toSocket + ") from local address " + fromAddress
 					+ " with timeout " + timeout);
 		}
-		Socket socket = new Socket();
-		try {
+		try (Socket socket = new Socket()) {
 			// port 0 = let the OS choose the port
 			SocketAddress bindP = new InetSocketAddress(fromAddress, 0);
 			// machine
@@ -300,43 +298,6 @@ private static boolean tryToConnect(InetAddress fromAddress, SocketAddress toSoc
 			}
 			return false;
 		}
-		finally {
-			socket.close();
-		}
-	}
-
-	/**
-	 * Find a non-occupied port.
-	 *
-	 * @return A non-occupied port.
-	 */
-	public static int getAvailablePort() {
-		for (int i = 0; i < 50; i++) {
-			ServerSocket serverSocket = null;
-			try {
-				serverSocket = new ServerSocket(0);
-				int port = serverSocket.getLocalPort();
-				if (port != 0) {
-					return port;
-				}
-			}
-			catch (IOException e) {
-				if (LOG.isDebugEnabled()) {
-					LOG.debug("Unable to allocate port " + e.getMessage(), e);
-				}
-			}
-			finally {
-				if (serverSocket != null) {
-					try {
-						serverSocket.close();
-					} catch (Throwable t) {
-						// ignored
-					}
-				}
-			}
-		}
-
-		throw new RuntimeException("Could not find a free permitted port on the machine.");
 	}
 
 	public static class LeaderConnectingAddressListener implements LeaderRetrievalListener {
@@ -408,7 +369,7 @@ public InetAddress findConnectingAddress(
 						}
 
 						do {
-							InetAddress address = NetUtils.findAddressUsingStrategy(strategy, targetAddress, logging);
+							InetAddress address = ConnectionUtils.findAddressUsingStrategy(strategy, targetAddress, logging);
 							if (address != null) {
 								return address;
 							}

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/LeaderRetrievalUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/LeaderRetrievalUtils.java
@@ -29,7 +29,7 @@
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalException;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
-import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.runtime.net.ConnectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Await;
@@ -149,7 +149,7 @@ public static LeaderConnectionInfo retrieveLeaderConnectionInfo(
 	public static InetAddress findConnectingAddress(
 			LeaderRetrievalService leaderRetrievalService,
 			FiniteDuration timeout) throws LeaderRetrievalException {
-		NetUtils.LeaderConnectingAddressListener listener = new NetUtils.LeaderConnectingAddressListener();
+		ConnectionUtils.LeaderConnectingAddressListener listener = new ConnectionUtils.LeaderConnectingAddressListener();
 
 		try {
 			leaderRetrievalService.start(listener);

diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -60,7 +60,7 @@ import org.apache.flink.runtime.memory.MemoryManager
 import org.apache.flink.runtime.messages.Messages._
 import org.apache.flink.runtime.messages.RegistrationMessages._
 import org.apache.flink.runtime.messages.TaskManagerMessages._
-import org.apache.flink.runtime.net.NetUtils
+import org.apache.flink.util.NetUtils
 import org.apache.flink.runtime.process.ProcessReaper
 import org.apache.flink.runtime.security.SecurityUtils
 import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/NetworkEnvironmentTest.java
@@ -26,10 +26,12 @@
 import org.apache.flink.runtime.io.disk.iomanager.IOManager;
 import org.apache.flink.runtime.io.network.buffer.BufferPool;
 import org.apache.flink.runtime.io.network.netty.NettyConfig;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.runtime.taskmanager.NetworkEnvironmentConfiguration;
 import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.Test;
+
 import scala.Some;
 import scala.Tuple2;
 import scala.concurrent.duration.FiniteDuration;

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyTestUtil.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/NettyTestUtil.java
@@ -19,8 +19,10 @@
 package org.apache.flink.runtime.io.network.netty;
 
 import io.netty.channel.Channel;
+
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.util.NetUtils;
+
 import scala.Tuple2;
 
 import java.net.InetAddress;
@@ -48,10 +50,7 @@ static NettyServer initServer(NettyConfig config, NettyProtocol protocol) throws
 			server.init(protocol);
 		}
 		catch (Exception e) {
-			if (server != null) {
-				server.shutdown();
-			}
-
+			server.shutdown();
 			throw e;
 		}
 
@@ -65,10 +64,7 @@ static NettyClient initClient(NettyConfig config, NettyProtocol protocol) throws
 			client.init(protocol);
 		}
 		catch (Exception e) {
-			if (client != null) {
-				client.shutdown();
-			}
-
+			client.shutdown();
 			throw e;
 		}
 

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerStartupTest.java
@@ -30,7 +30,7 @@
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.StreamingMode;
-import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.util.NetUtils;
 
 import org.apache.flink.util.OperatingSystem;
 import org.junit.After;

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java
@@ -33,11 +33,13 @@
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.runtime.messages.JobManagerMessages;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.runtime.util.LeaderRetrievalUtils;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+
 import scala.Tuple2;
 import scala.concurrent.Await;
 import scala.concurrent.Future;

diff --git a/...pache/flink/runtime/net/NetUtilsTest.java → ...link/runtime/net/ConnectionUtilsTest.java b/...pache/flink/runtime/net/NetUtilsTest.java → ...link/runtime/net/ConnectionUtilsTest.java
@@ -28,13 +28,13 @@
 /**
  * Tests for the network utilities.
  */
-public class NetUtilsTest {
+public class ConnectionUtilsTest {
 
 	@Test
 	public void testFindConnectableAddress() {
 		int unusedPort;
 		try {
-			unusedPort = NetUtils.getAvailablePort();
+			unusedPort = org.apache.flink.util.NetUtils.getAvailablePort();
 		}
 		catch (Throwable t) {
 			// if this system cannot find an available port,
@@ -47,7 +47,7 @@ public void testFindConnectableAddress() {
 			InetSocketAddress unreachable = new InetSocketAddress("localhost", unusedPort);
 
 			final long start = System.currentTimeMillis();
-			InetAddress add = NetUtils.findConnectingAddress(unreachable, 2000, 400);
+			InetAddress add = ConnectionUtils.findConnectingAddress(unreachable, 2000, 400);
 
 			// check that it did not take forever
 			assertTrue(System.currentTimeMillis() - start < (OperatingSystem.isWindows() ? 30000 : 8000));

diff --git a/...ime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerProcessReapingTest.java b/...ime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerProcessReapingTest.java
@@ -26,10 +26,10 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.StreamingMode;
 import org.apache.flink.runtime.akka.AkkaUtils;
-
 import org.apache.flink.runtime.jobmanager.JobManager;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.Test;
 
 import scala.Some;

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java
@@ -57,10 +57,10 @@
 import org.apache.flink.runtime.messages.TaskMessages.PartitionState;
 import org.apache.flink.runtime.messages.TaskMessages.SubmitTask;
 import org.apache.flink.runtime.messages.TaskMessages.TaskOperationResult;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages;
-
 import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;

diff --git a/...runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerConnectionTest.scala b/...runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerConnectionTest.scala
@@ -25,7 +25,7 @@ import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
 
 import org.apache.flink.configuration.{ConfigConstants, Configuration}
 import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.net.NetUtils
+import org.apache.flink.util.NetUtils
 import org.junit.Assert._
 import org.junit.Test
 

diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -28,10 +28,9 @@ import org.apache.flink.runtime.StreamingMode
 import org.apache.flink.runtime.jobmanager.JobManager
 import org.apache.flink.runtime.leaderelection.LeaderElectionService
 import org.apache.flink.runtime.minicluster.FlinkMiniCluster
-import org.apache.flink.runtime.net.NetUtils
+import org.apache.flink.util.NetUtils
 import org.apache.flink.runtime.taskmanager.TaskManager
 import org.apache.flink.runtime.testingUtils.TestingMessages.Alive
-import org.apache.flink.runtime.webmonitor.WebMonitor
 
 import scala.concurrent.{Await, Future}
 

diff --git a/...nector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/...nector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
@@ -33,11 +33,11 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.StreamingMode;
 import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.internals.ZooKeeperStringSerializer;
 import org.apache.flink.streaming.connectors.kafka.testutils.SuccessException;
 import org.apache.flink.test.util.ForkableFlinkMiniCluster;
+import org.apache.flink.util.NetUtils;
 import org.apache.flink.util.TestLogger;
 
 import org.apache.kafka.common.PartitionInfo;

diff --git a/...nk-streaming-core/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java b/...nk-streaming-core/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java
@@ -19,10 +19,11 @@
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.streaming.util.serialization.DeserializationSchema;
 import org.apache.flink.streaming.util.serialization.SerializationSchema;
 import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.Assert;
 
 import java.io.BufferedReader;

diff --git a/...streaming-core/src/test/java/org/apache/flink/streaming/util/SocketProgramITCaseBase.java b/...streaming-core/src/test/java/org/apache/flink/streaming/util/SocketProgramITCaseBase.java
@@ -17,8 +17,9 @@
 
 package org.apache.flink.streaming.util;
 
-import org.apache.flink.runtime.net.NetUtils;
 import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.util.NetUtils;
+
 import org.junit.Assert;
 
 import java.io.PrintWriter;