diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index fcbef4c220..f57f2b76e2 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -140,6 +140,7 @@ public abstract class AbstractCluster implements ICluster nodeIdTopology; private final Consumer configUpdater; private final int broadcastPort; + private final Map portMap; // mutated by starting/stopping a node private final List instances; @@ -163,6 +164,7 @@ public abstract class AbstractCluster implements ICluster { private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces; + private boolean dynamicPortAllocation = false; { // Indicate that we are running in the in-jvm dtest environment @@ -177,10 +179,32 @@ public abstract class AbstractCluster implements ICluster implements ICluster() : null; int generation = GENERATION.incrementAndGet(); for (int i = 0; i < builder.getNodeCount(); ++i) @@ -431,7 +456,7 @@ public abstract class AbstractCluster implements ICluster portMap) + { + String ipAdress = "127.0." + subnet + ".1"; return new INodeProvisionStrategy() { + @Override public String seedIp() { - return "127.0." + subnet + ".1"; + return ipAdress; } + @Override public int seedPort() { - return 7012; + return storagePort(1); } + @Override public String ipAddress(int nodeNum) { - return "127.0." + subnet + ".1"; + return ipAdress; } + @Override public int storagePort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("storagePort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 7011 + nodeNum)); + } return 7011 + nodeNum; } + @Override public int nativeTransportPort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("nativeTransportPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 9041 + nodeNum)); + } return 9041 + nodeNum; } + @Override public int jmxPort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("jmxPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(seedIp(), 7199 + nodeNum)); + } return 7199 + nodeNum; } }; @@ -62,49 +87,81 @@ public interface INodeProvisionStrategy }, MultipleNetworkInterfaces { - INodeProvisionStrategy create(int subnet) { - String ipPrefix = "127.0." + subnet + "."; + @Override + INodeProvisionStrategy create(int subnet, @Nullable Map portMap) + { + String ipPrefix = "127.0." + subnet + '.'; return new INodeProvisionStrategy() { + + @Override public String seedIp() { - return ipPrefix + "1"; + return ipPrefix + '1'; } + @Override public int seedPort() { - return 7012; + return storagePort(1); } + @Override public String ipAddress(int nodeNum) { return ipPrefix + nodeNum; } + @Override public int storagePort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("storagePort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 7012)); + } return 7012; } + @Override public int nativeTransportPort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("nativeTransportPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 9042)); + } return 9042; } + @Override public int jmxPort(int nodeNum) { + if (portMap != null) + { + return portMap.computeIfAbsent("jmxPort@node" + nodeNum, key -> SocketUtils.findAvailablePort(ipAddress(nodeNum), 7199)); + } return 7199; } }; } }; - abstract INodeProvisionStrategy create(int subnet); + + INodeProvisionStrategy create(int subnet) + { + return create(subnet, null); + } + + abstract INodeProvisionStrategy create(int subnet, @Nullable Map portMap); } - abstract String seedIp(); - abstract int seedPort(); - abstract String ipAddress(int nodeNum); - abstract int storagePort(int nodeNum); - abstract int nativeTransportPort(int nodeNum); - abstract int jmxPort(int nodeNum); + String seedIp(); + + int seedPort(); + + String ipAddress(int nodeNum); + + int storagePort(int nodeNum); + + int nativeTransportPort(int nodeNum); + + int jmxPort(int nodeNum); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 06229a0c01..d1714a33bf 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -110,7 +110,7 @@ public class InstanceConfig implements IInstanceConfig .set("native_transport_port", native_transport_port) .set("endpoint_snitch", DistributedTestSnitch.class.getName()) .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(), - Collections.singletonMap("seeds", seedIp + ":" + seedPort))) + Collections.singletonMap("seeds", seedIp + ':' + seedPort))) // required settings for dtest functionality .set("diagnostic_events_enabled", true) .set("auto_bootstrap", false) diff --git a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java index ea1efa8902..b6b0eba2f6 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/jmx/JMXFeatureTest.java @@ -72,6 +72,7 @@ public class JMXFeatureTest extends TestBaseImpl for (int i = 0; i < iterations; i++) { try (Cluster cluster = Cluster.build(2) + .withDynamicPortAllocation(true) .withNodeProvisionStrategy(provisionStrategy) .withConfig(c -> c.with(Feature.values())).start()) { @@ -136,7 +137,7 @@ public class JMXFeatureTest extends TestBaseImpl // to check that we are actually connecting to the correct instance String defaultDomain = mbsc.getDefaultDomain(); instancesContacted.add(defaultDomain); - Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ":" + config.jmxPort())); + Assert.assertThat(defaultDomain, startsWith(JMXUtil.getJmxHost(config) + ':' + config.jmxPort())); } catch (Throwable t) { diff --git a/test/unit/org/apache/cassandra/net/SocketUtils.java b/test/unit/org/apache/cassandra/net/SocketUtils.java index a0a149029b..6a6cd559e2 100644 --- a/test/unit/org/apache/cassandra/net/SocketUtils.java +++ b/test/unit/org/apache/cassandra/net/SocketUtils.java @@ -19,39 +19,54 @@ package org.apache.cassandra.net; import java.io.IOException; +import java.net.InetAddress; import java.net.ServerSocket; +import java.net.UnknownHostException; import com.google.common.base.Throwables; public class SocketUtils { - public static synchronized int findAvailablePort() throws RuntimeException + /** + * Returns an available port for the given {@code bindAddress}. When an {@link IOException} occurs when opening a + * socket or if a {@link SecurityException} is raised because a manager exists and its checkListen method does + * not allow the operation, the {@code fallbackPort} is returned. + * + * @param bindAddress the ip address for the interface where we need an available port number + * @param fallbackPort a port to return in case {@link SecurityException} or {@link IOException} is encountered + * @return an available port the given {@code bindAddress} when succeeds, otherwise the {@code fallbackPort} + * @throws RuntimeException if no IP address for the {@code bindAddress} could be found + */ + public static synchronized int findAvailablePort(String bindAddress, int fallbackPort) throws RuntimeException { - ServerSocket ss = null; try { - // let the system pick an ephemeral port - ss = new ServerSocket(0); - ss.setReuseAddress(true); - return ss.getLocalPort(); + return findAvailablePort(InetAddress.getByName(bindAddress), fallbackPort); } - catch (IOException e) + catch (UnknownHostException e) { - throw Throwables.propagate(e); - } - finally - { - if (ss != null) - { - try - { - ss.close(); - } - catch (IOException e) - { - Throwables.propagate(e); - } - } + throw new RuntimeException(e); } } -} \ No newline at end of file + + /** + * Returns an available port for the given {@code bindAddress}. When an {@link IOException} occurs when opening a + * socket or if a {@link SecurityException} is raised because a manager exists and its checkListen method does + * not allow the operation, the {@code fallbackPort} is returned. + * + * @param bindAddress the ip address for the interface where we need an available port number + * @param fallbackPort a port to return in case {@link SecurityException} or {@link IOException} is encountered + * @return an available port the given {@code bindAddress} when succeeds, otherwise the {@code fallbackPort} + */ + public static synchronized int findAvailablePort(InetAddress bindAddress, int fallbackPort) + { + try (ServerSocket socket = new ServerSocket(0, 50, bindAddress)) + { + return socket.getLocalPort(); + } + catch (SecurityException | IOException exception) + { + return fallbackPort; + } + } +}