[flink] branch master updated: [FLINK-10580] Harden BootstrapTool#startActorSystem

[tr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 7dbe4bb  [FLINK-10580] Harden BootstrapTool#startActorSystem
7dbe4bb is described below

commit 7dbe4bb194f79cd996b46c36037b512e93b1ef07
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Wed Oct 17 13:35:51 2018 +0200

    [FLINK-10580] Harden BootstrapTool#startActorSystem
    
    Instead of opening a socket to check whether a given port is free we simply start an
    ActorSystem with it an check whether it can bind to this port. This also solves the problem
    that a 0 port get resolved to a specific port which might get taken between closing the
    test socket and starting the ActorSystem.
---
 .../runtime/clusterframework/BootstrapTools.java   | 15 +-----
 .../clusterframework/BootstrapToolsTest.java       | 57 +++++++++++++++++++++-
 2 files changed, 57 insertions(+), 15 deletions(-)

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
index d95034d..3a2d2c5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
@@ -55,7 +55,6 @@ import java.io.FileWriter;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.net.BindException;
-import java.net.ServerSocket;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -157,19 +156,7 @@ public class BootstrapTools {
 		}
 
 		while (portsIterator.hasNext()) {
-			// first, we check if the port is available by opening a socket
-			// if the actor system fails to start on the port, we try further
-			ServerSocket availableSocket = NetUtils.createSocketFromPorts(portsIterator, ServerSocket::new);
-
-			int port;
-			if (availableSocket == null) {
-				throw new BindException("Unable to allocate further port in port range: " + portRangeDefinition);
-			} else {
-				port = availableSocket.getLocalPort();
-				try {
-					availableSocket.close();
-				} catch (IOException ignored) {}
-			}
+			final int port = portsIterator.next();
 
 			try {
 				return startActorSystem(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
index d1f32cf..c01f1aa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
@@ -21,10 +21,26 @@ package org.apache.flink.runtime.clusterframework;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.function.CheckedSupplier;
 
+import akka.actor.ActorSystem;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -33,7 +49,9 @@ import static org.junit.Assert.assertNull;
 /**
  * Tests for {@link BootstrapToolsTest}.
  */
-public class BootstrapToolsTest {
+public class BootstrapToolsTest extends TestLogger {
+
+	private static final Logger LOG = LoggerFactory.getLogger(BootstrapToolsTest.class);
 
 	@Test
 	public void testSubstituteConfigKey() {
@@ -305,4 +323,41 @@ public class BootstrapToolsTest {
 		BootstrapTools.updateTmpDirectoriesInConfiguration(config, null);
 		assertEquals(config.getString(CoreOptions.TMP_DIRS), CoreOptions.TMP_DIRS.defaultValue());
 	}
+
+	/**
+	 * Tests that we can concurrently create two {@link ActorSystem} without port conflicts.
+	 * This effectively tests that we don't open a socket to check for a ports availability.
+	 * See FLINK-10580 for more details.
+	 */
+	@Test
+	public void testConcurrentActorSystemCreation() throws Exception {
+		final int concurrentCreations = 10;
+		final ExecutorService executorService = Executors.newFixedThreadPool(concurrentCreations);
+		final CyclicBarrier cyclicBarrier = new CyclicBarrier(concurrentCreations);
+
+		try {
+			final List<CompletableFuture<Void>> actorSystemFutures = IntStream.range(0, concurrentCreations)
+				.mapToObj(
+					ignored ->
+						CompletableFuture.supplyAsync(
+							CheckedSupplier.unchecked(() -> {
+								cyclicBarrier.await();
+
+								return BootstrapTools.startActorSystem(
+									new Configuration(),
+									"localhost",
+									"0",
+									LOG);
+							})))
+				.map(
+					// terminate ActorSystems
+					actorSystemFuture ->
+						actorSystemFuture.thenCompose(AkkaUtils::terminateActorSystem)
+				).collect(Collectors.toList());
+
+			FutureUtils.completeAll(actorSystemFutures).get();
+		} finally {
+			ExecutorUtils.gracefulShutdown(10000L, TimeUnit.MILLISECONDS, executorService);
+		}
+	}
 }