Upgrading Guava to 27, and to java-driver 3.6.0 (from 3.4.0-SNAPSHOT), plus refactoring to remove nativePort argument for NativeSSTableLoaderClient constructor

Patch by Sumanth Pasupuleti; reviewed by Michael Semb Wever for CASSANDRA-14655

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0-alpha2
+ * Upgrade Guava to 27, and to java-driver 3.6.0 (from 3.4.0-SNAPSHOT) (CASSANDRA-14655)
  * Extract an AbstractCompactionController to allow for custom implementations (CASSANDRA-15286)
  * Move chronicle-core version from snapshot to stable, and include carrotsearch in generated pom.xml (CASSANDRA-15321)
  * Untangle RepairMessage sub-hierarchy of messages, use new messaging (more) correctly (CASSANDRA-15163)

build.xml

@@ -487,7 +487,7 @@
           <dependency groupId="org.lz4" artifactId="lz4-java" version="1.4.0"/>
           <dependency groupId="com.ning" artifactId="compress-lzf" version="0.8.4"/>
           <dependency groupId="com.github.luben" artifactId="zstd-jni" version="1.3.8-5"/>
-          <dependency groupId="com.google.guava" artifactId="guava" version="23.3-jre"/>
+          <dependency groupId="com.google.guava" artifactId="guava" version="27.0-jre"/>
           <dependency groupId="org.hdrhistogram" artifactId="HdrHistogram" version="2.1.9"/>
           <dependency groupId="commons-cli" artifactId="commons-cli" version="1.1"/>
           <dependency groupId="commons-codec" artifactId="commons-codec" version="1.9"/>
@@ -557,15 +557,13 @@
           <dependency groupId="net.openhft" artifactId="chronicle-threads" version="${chronicle-threads.version}"/>
           <dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" />
           <dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />
-	  <!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
+          <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="3.6.0" classifier="shaded">
-          <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="3.4.0-SNAPSHOT" classifier="shaded">
             <exclusion groupId="io.netty" artifactId="netty-buffer"/>
             <exclusion groupId="io.netty" artifactId="netty-codec"/>
             <exclusion groupId="io.netty" artifactId="netty-handler"/>
             <exclusion groupId="io.netty" artifactId="netty-transport"/>
             <exclusion groupId="org.slf4j" artifactId="slf4j-api"/>
           </dependency>
-	  -->
           <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj" version="${ecj.version}" />
           <dependency groupId="org.caffinitas.ohc" artifactId="ohc-core" version="${ohc.version}">
             <exclusion groupId="org.slf4j" artifactId="slf4j-api"/>
@@ -647,9 +645,7 @@
         <dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster"/>
         <dependency groupId="com.google.code.findbugs" artifactId="jsr305"/>
         <dependency groupId="org.antlr" artifactId="antlr"/>
-	<!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
         <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded"/>
-	-->
         <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
         <dependency groupId="org.caffinitas.ohc" artifactId="ohc-core"/>
         <dependency groupId="org.caffinitas.ohc" artifactId="ohc-core-j8"/>
@@ -666,9 +662,7 @@
                 artifactId="cassandra-parent"
                 version="${version}"/>
         <dependency groupId="junit" artifactId="junit"/>
-	<!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
         <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded"/>
-	-->
         <dependency groupId="io.netty" artifactId="netty-all"/>
         <dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
         <dependency groupId="org.caffinitas.ohc" artifactId="ohc-core"/>
@@ -743,14 +737,12 @@
         <dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster" optional="true"/>
 
         <!-- don't need the Java Driver to run, but if you use the hadoop stuff or UDFs -->
-	<!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
         <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded" optional="true">
           <exclusion groupId="io.netty" artifactId="netty-buffer"/>
           <exclusion groupId="io.netty" artifactId="netty-codec"/>
           <exclusion groupId="io.netty" artifactId="netty-handler"/>
           <exclusion groupId="io.netty" artifactId="netty-transport"/>
         </dependency>
-	-->
 
         <!-- don't need jna to run, but nice to have -->
         <dependency groupId="net.java.dev.jna" artifactId="jna"/>

src/java/org/apache/cassandra/audit/AuditLogManager.java

@@ -37,7 +37,6 @@ import org.apache.cassandra.cql3.statements.BatchStatement;
 import org.apache.cassandra.exceptions.AuthenticationException;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.UnauthorizedException;
-import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.utils.FBUtilities;
 

src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java

@@ -287,12 +287,10 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
         public ExternalClient(Configuration conf)
         {
             super(resolveHostAddresses(conf),
-                  CqlConfigHelper.getOutputNativePort(conf),
                   ConfigHelper.getOutputInitialPort(conf),
                   ConfigHelper.getOutputKeyspaceUserName(conf),
                   ConfigHelper.getOutputKeyspacePassword(conf),
-                  CqlConfigHelper.getSSLOptions(conf).orNull(),
+                  CqlConfigHelper.getSSLOptions(conf).orNull());
-                  CqlConfigHelper.getAllowServerPortDiscovery(conf));
         }
 
         private static Collection<InetSocketAddress> resolveHostAddresses(Configuration conf)

src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java

@@ -88,7 +88,6 @@ public class CqlConfigHelper
 
     private static final String OUTPUT_CQL = "cassandra.output.cql";
     private static final String OUTPUT_NATIVE_PORT = "cassandra.output.native.port";
-    private static final String ALLOW_SERVER_PORT_DISCOVERY = "cassandra.allowserverportdiscovery";
 
     /**
      * Set the CQL columns for the input of this job.
@@ -652,15 +651,4 @@ public class CqlConfigHelper
                  new SecureRandom());
         return ctx;
     }
-
-    public static void setAllowServerPortDiscovery(Configuration conf, boolean allowServerPortDiscovery)
-    {
-        conf.set(ALLOW_SERVER_PORT_DISCOVERY, Boolean.toString(allowServerPortDiscovery));
-    }
-
-    public static boolean getAllowServerPortDiscovery(Configuration conf)
-    {
-        return Boolean.parseBoolean(conf.get(ALLOW_SERVER_PORT_DISCOVERY, "false"));
-    }
-
 }

src/java/org/apache/cassandra/index/SecondaryIndexManager.java

@@ -487,7 +487,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
                                        builtIndexes.addAll(groupedIndexes);
                                        build.set(o);
                                    }
-                               });
+                               }, MoreExecutors.directExecutor());
                                futures.add(build);
                            });
 
@@ -1444,7 +1444,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
         if (null != task)
         {
             ListenableFuture<?> f = blockingExecutor.submit(task);
-            if (callback != null) Futures.addCallback(f, callback);
+            if (callback != null) Futures.addCallback(f, callback, MoreExecutors.directExecutor());
             FBUtilities.waitOnFuture(f);
         }
     }
@@ -1464,7 +1464,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
                              if (null != task)
                              {
                                  ListenableFuture<?> f = blockingExecutor.submit(task);
-                                 if (callback != null) Futures.addCallback(f, callback);
+                                 if (callback != null) Futures.addCallback(f, callback, MoreExecutors.directExecutor());
                                  waitFor.add(f);
                              }
                          });

src/java/org/apache/cassandra/io/sstable/format/SSTableFormat.java

@@ -57,7 +57,7 @@ public interface SSTableFormat
         {
             //Since format comes right after generation
             //we disallow formats with numeric names
-            assert !CharMatcher.DIGIT.matchesAllOf(name);
+            assert !CharMatcher.digit().matchesAllOf(name);
 
             this.name = name;
             this.info = info;

src/java/org/apache/cassandra/repair/RepairJob.java

@@ -391,7 +391,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
 
                 // failure is handled at root of job chain
                 public void onFailure(Throwable t) {}
-            });
+            }, MoreExecutors.directExecutor());
             currentTask = nextTask;
         }
         // start running tasks
@@ -448,7 +448,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
 
                     // failure is handled at root of job chain
                     public void onFailure(Throwable t) {}
-                });
+                }, MoreExecutors.directExecutor());
                 currentTask = nextTask;
             }
             // start running tasks

src/java/org/apache/cassandra/repair/RepairRunnable.java

@@ -388,7 +388,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
         {
             ranges.addAll(range);
         }
-        Futures.addCallback(repairResult, new RepairCompleteCallback(parentSession, ranges, startTime, traceState, hasFailure, executor));
+        Futures.addCallback(repairResult, new RepairCompleteCallback(parentSession, ranges, startTime, traceState, hasFailure, executor), MoreExecutors.directExecutor());
     }
 
     private void previewRepair(UUID parentSession,
@@ -463,7 +463,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
                 executor.shutdownNow();
                 return message;
             }
-        });
+        }, MoreExecutors.directExecutor());
     }
 
     private ListenableFuture<List<RepairSessionResult>> submitRepairSessions(UUID parentSession,
@@ -493,7 +493,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
                                                                                      cfnames);
             if (session == null)
                 continue;
-            Futures.addCallback(session, new RepairSessionCallback(session));
+            Futures.addCallback(session, new RepairSessionCallback(session), MoreExecutors.directExecutor());
             futures.add(session);
         }
         return Futures.successfulAsList(futures);

src/java/org/apache/cassandra/repair/RepairSession.java

@@ -344,7 +344,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
                 Tracing.traceRepair("Session completed with the following error: {}", t);
                 forceShutdown(t);
             }
-        });
+        }, MoreExecutors.directExecutor());
     }
 
     public void terminate()

src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java

@@ -382,7 +382,7 @@ public class CoordinatorSession extends ConsistentSession
                     resultFuture.setException(t);
                 }
             }
-        });
+        }, MoreExecutors.directExecutor());
 
         return resultFuture;
     }

src/java/org/apache/cassandra/repair/consistent/LocalSessions.java

@@ -48,6 +48,7 @@ import com.google.common.primitives.Ints;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 
 import org.apache.cassandra.locator.RangesAtEndpoint;
 import org.slf4j.Logger;
@@ -666,7 +667,7 @@ public class LocalSessions
                     executor.shutdown();
                 }
             }
-        });
+        }, MoreExecutors.directExecutor());
     }
 
     public void maybeSetRepairing(UUID sessionID)

src/java/org/apache/cassandra/service/StorageService.java

@@ -1584,7 +1584,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
             {
                 logger.warn("Error during bootstrap.", e);
             }
-        });
+        }, MoreExecutors.directExecutor());
         try
         {
             bootstrapStream.get();
@@ -1679,7 +1679,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
                     progressSupport.progress("bootstrap", new ProgressEvent(ProgressEventType.ERROR, 1, 1, message));
                     progressSupport.progress("bootstrap", new ProgressEvent(ProgressEventType.COMPLETE, 1, 1, "Resume bootstrap complete"));
                 }
-            });
+            }, MoreExecutors.directExecutor());
             return true;
         }
         else
@@ -2958,7 +2958,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
                 // We still want to send the notification
                 sendReplicationNotification(notifyEndpoint);
             }
-        });
+        }, MoreExecutors.directExecutor());
     }
 
     /**

src/java/org/apache/cassandra/streaming/StreamResultFuture.java

@@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 
 import com.google.common.util.concurrent.AbstractFuture;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -145,7 +146,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
 
     public void addEventListener(StreamEventHandler listener)
     {
-        Futures.addCallback(this, listener);
+        Futures.addCallback(this, listener, MoreExecutors.directExecutor());
         eventListeners.add(listener);
     }
 

src/java/org/apache/cassandra/tools/BulkLoader.java

@@ -57,13 +57,11 @@ public class BulkLoader
                 options.directory.getAbsoluteFile(),
                 new ExternalClient(
                         options.hosts,
-                        options.nativePort,
                         options.storagePort,
                         options.authProvider,
                         options.sslStoragePort,
                         options.serverEncOptions,
-                        buildSSLOptions(options.clientEncOptions),
+                        buildSSLOptions(options.clientEncOptions)),
-                        options.allowServerPortDiscovery),
                         handler,
                         options.connectionsPerHost,
                         options.targetKeyspace);
@@ -275,15 +273,13 @@ public class BulkLoader
         private final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
 
         public ExternalClient(Set<InetSocketAddress> hosts,
-                              int nativePort,
                               int storagePort,
                               AuthProvider authProvider,
                               int sslStoragePort,
                               EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions,
-                              SSLOptions sslOptions,
+                              SSLOptions sslOptions)
-                              boolean allowServerPortDiscovery)
         {
-            super(hosts, nativePort, storagePort, authProvider, sslOptions, allowServerPortDiscovery);
+            super(hosts, storagePort, authProvider, sslOptions);
             this.sslStoragePort = sslStoragePort;
             serverEncOptions = serverEncryptionOptions;
         }

src/java/org/apache/cassandra/tools/LoaderOptions.java

@@ -59,7 +59,6 @@ public class LoaderOptions
     public static final String THROTTLE_MBITS = "throttle";
     public static final String INTER_DC_THROTTLE_MBITS = "inter-dc-throttle";
     public static final String TOOL_NAME = "sstableloader";
-    public static final String ALLOW_SERVER_PORT_DISCOVERY_OPTION = "server-port-discovery";
     public static final String TARGET_KEYSPACE = "target-keyspace";
 
     /* client encryption options */
@@ -89,7 +88,6 @@ public class LoaderOptions
     public final EncryptionOptions.ServerEncryptionOptions serverEncOptions;
     public final Set<InetSocketAddress> hosts;
     public final Set<InetAddressAndPort> ignores;
-    public final boolean allowServerPortDiscovery;
     public final String targetKeyspace;
 
     LoaderOptions(Builder builder)
@@ -110,7 +108,6 @@ public class LoaderOptions
         connectionsPerHost = builder.connectionsPerHost;
         serverEncOptions = builder.serverEncOptions;
         hosts = builder.hosts;
-        allowServerPortDiscovery = builder.allowServerPortDiscovery;
         ignores = builder.ignores;
         targetKeyspace = builder.targetKeyspace;
     }
@@ -137,7 +134,6 @@ public class LoaderOptions
         Set<InetAddress> ignoresArg = new HashSet<>();
         Set<InetSocketAddress> hosts = new HashSet<>();
         Set<InetAddressAndPort> ignores = new HashSet<>();
-        boolean allowServerPortDiscovery;
         String targetKeyspace;
 
         Builder()
@@ -307,12 +303,6 @@ public class LoaderOptions
             return this;
         }
 
-        public Builder allowServerPortDiscovery(boolean allowServerPortDiscovery)
-        {
-            this.allowServerPortDiscovery = allowServerPortDiscovery;
-            return this;
-        }
-
         public Builder parseArgs(String cmdArgs[])
         {
             CommandLineParser parser = new GnuParser();
@@ -359,7 +349,6 @@ public class LoaderOptions
 
                 verbose = cmd.hasOption(VERBOSE_OPTION);
                 noProgress = cmd.hasOption(NOPROGRESS_OPTION);
-                allowServerPortDiscovery = cmd.hasOption(ALLOW_SERVER_PORT_DISCOVERY_OPTION);
 
                 if (cmd.hasOption(USER_OPTION))
                 {
@@ -376,48 +365,6 @@ public class LoaderOptions
                     authProviderName = cmd.getOptionValue(AUTH_PROVIDER_OPTION);
                 }
 
-                if (cmd.hasOption(INITIAL_HOST_ADDRESS_OPTION))
-                {
-                    String[] nodes = cmd.getOptionValue(INITIAL_HOST_ADDRESS_OPTION).split(",");
-                    try
-                    {
-                        for (String node : nodes)
-                        {
-                            HostAndPort hap = HostAndPort.fromString(node);
-                            hosts.add(new InetSocketAddress(InetAddress.getByName(hap.getHost()), hap.getPortOrDefault(nativePort)));
-                        }
-                    } catch (UnknownHostException e)
-                    {
-                        errorMsg("Unknown host: " + e.getMessage(), options);
-                    }
-
-                } else
-                {
-                    System.err.println("Initial hosts must be specified (-d)");
-                    printUsage(options);
-                    System.exit(1);
-                }
-
-                if (cmd.hasOption(IGNORE_NODES_OPTION))
-                {
-                    String[] nodes = cmd.getOptionValue(IGNORE_NODES_OPTION).split(",");
-                    try
-                    {
-                        for (String node : nodes)
-                        {
-                            ignores.add(InetAddressAndPort.getByNameOverrideDefaults(node.trim(), storagePort));
-                        }
-                    } catch (UnknownHostException e)
-                    {
-                        errorMsg("Unknown host: " + e.getMessage(), options);
-                    }
-                }
-
-                if (cmd.hasOption(CONNECTIONS_PER_HOST))
-                {
-                    connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST));
-                }
-
                 // try to load config file first, so that values can be
                 // rewritten with other option values.
                 // otherwise use default config.
@@ -443,10 +390,54 @@ public class LoaderOptions
                     nativePort = Integer.parseInt(cmd.getOptionValue(NATIVE_PORT_OPTION));
                 else
                     nativePort = config.native_transport_port;
+
+                if (cmd.hasOption(INITIAL_HOST_ADDRESS_OPTION))
+                {
+                    String[] nodes = cmd.getOptionValue(INITIAL_HOST_ADDRESS_OPTION).split(",");
+                    try
+                    {
+                        for (String node : nodes)
+                        {
+                            HostAndPort hap = HostAndPort.fromString(node);
+                            hosts.add(new InetSocketAddress(InetAddress.getByName(hap.getHost()), hap.getPortOrDefault(nativePort)));
+                        }
+                    } catch (UnknownHostException e)
+                    {
+                        errorMsg("Unknown host: " + e.getMessage(), options);
+                    }
+
+                } else
+                {
+                    System.err.println("Initial hosts must be specified (-d)");
+                    printUsage(options);
+                    System.exit(1);
+                }
+
                 if (cmd.hasOption(STORAGE_PORT_OPTION))
                     storagePort = Integer.parseInt(cmd.getOptionValue(STORAGE_PORT_OPTION));
                 else
                     storagePort = config.storage_port;
+
+                if (cmd.hasOption(IGNORE_NODES_OPTION))
+                {
+                    String[] nodes = cmd.getOptionValue(IGNORE_NODES_OPTION).split(",");
+                    try
+                    {
+                        for (String node : nodes)
+                        {
+                            ignores.add(InetAddressAndPort.getByNameOverrideDefaults(node.trim(), storagePort));
+                        }
+                    } catch (UnknownHostException e)
+                    {
+                        errorMsg("Unknown host: " + e.getMessage(), options);
+                    }
+                }
+
+                if (cmd.hasOption(CONNECTIONS_PER_HOST))
+                {
+                    connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST));
+                }
+
                 if (cmd.hasOption(SSL_STORAGE_PORT_OPTION))
                     sslStoragePort = Integer.parseInt(cmd.getOptionValue(SSL_STORAGE_PORT_OPTION));
                 else
@@ -626,7 +617,6 @@ public class LoaderOptions
         options.addOption("st", SSL_STORE_TYPE, "STORE-TYPE", "Client SSL: type of store");
         options.addOption("ciphers", SSL_CIPHER_SUITES, "CIPHER-SUITES", "Client SSL: comma-separated list of encryption suites to use");
         options.addOption("f", CONFIG_PATH, "path to config file", "cassandra.yaml file path for streaming throughput and client/server SSL.");
-        options.addOption("spd", ALLOW_SERVER_PORT_DISCOVERY_OPTION, "allow server port discovery", "Use ports published by server to decide how to connect. With SSL requires StartTLS to be used.");
         options.addOption("k", TARGET_KEYSPACE, "target keyspace name", "target keyspace name");
         return options;
     }

src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java

@@ -46,37 +46,28 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client
 {
     protected final Map<String, TableMetadataRef> tables;
     private final Collection<InetSocketAddress> hosts;
-    private final int port;
     private final int storagePort;
     private final AuthProvider authProvider;
     private final SSLOptions sslOptions;
-    private final boolean allowServerPortDiscovery;
 
-
+    public NativeSSTableLoaderClient(Collection<InetSocketAddress> hosts, int storagePort, String username, String password, SSLOptions sslOptions)
-    public NativeSSTableLoaderClient(Collection<InetSocketAddress> hosts, int nativePort, int storagePort, String username, String password, SSLOptions sslOptions, boolean allowServerPortDiscovery)
     {
-        this(hosts, nativePort, storagePort, new PlainTextAuthProvider(username, password), sslOptions, allowServerPortDiscovery);
+        this(hosts, storagePort, new PlainTextAuthProvider(username, password), sslOptions);
     }
 
-    public NativeSSTableLoaderClient(Collection<InetSocketAddress> hosts, int nativePort, int storagePort, AuthProvider authProvider, SSLOptions sslOptions, boolean allowServerPortDiscovery)
+    public NativeSSTableLoaderClient(Collection<InetSocketAddress> hosts, int storagePort, AuthProvider authProvider, SSLOptions sslOptions)
     {
         super();
         this.tables = new HashMap<>();
         this.hosts = hosts;
-        this.port = nativePort;
         this.authProvider = authProvider;
         this.sslOptions = sslOptions;
-        this.allowServerPortDiscovery = allowServerPortDiscovery;
         this.storagePort = storagePort;
     }
 
     public void init(String keyspace)
     {
-        Set<InetAddress> hostAddresses = hosts.stream().map(host -> host.getAddress()).collect(Collectors.toSet());
+        Cluster.Builder builder = Cluster.builder().addContactPointsWithPorts(hosts).allowBetaProtocolVersion();
-        Cluster.Builder builder = Cluster.builder().addContactPoints(hostAddresses).withPort(port).allowBetaProtocolVersion();
-
-        if (allowServerPortDiscovery)
-            builder = builder.allowServerPortDiscovery();
 
         if (sslOptions != null)
             builder.withSSL(sslOptions);
@@ -100,15 +91,9 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client
                                                  tokenFactory.fromString(tokenRange.getEnd().getValue().toString()));
                 for (Host endpoint : endpoints)
                 {
-                    int portToUse;
+                    int broadcastPort = endpoint.getBroadcastSocketAddress().getPort();
-                    if (allowServerPortDiscovery)
+                    // use port from broadcast address if set.
-                    {
+                    int portToUse = broadcastPort != 0 ? broadcastPort : storagePort;
-                        portToUse = endpoint.getBroadcastAddressOptPort().portOrElse(storagePort);
-                    }
-                    else
-                    {
-                        portToUse = storagePort;
-                    }
                     addRangeForEndpoint(range, InetAddressAndPort.getByNameOverrideDefaults(endpoint.getAddress().getHostAddress(), portToUse));
                 }
             }

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -47,6 +47,8 @@ import com.datastax.driver.core.ResultSet;
 
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.db.virtual.VirtualKeyspaceRegistry;
+import org.apache.cassandra.db.virtual.VirtualSchemaKeyspace;
 import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.config.EncryptionOptions;
 import org.apache.cassandra.locator.InetAddressAndPort;
@@ -401,6 +403,8 @@ public abstract class CQLTester
             return;
 
         SystemKeyspace.finishStartup();
+        VirtualKeyspaceRegistry.instance.register(VirtualSchemaKeyspace.instance);
+
         StorageService.instance.initServer();
         SchemaLoader.startGossiper();
 

test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java

@@ -552,7 +552,7 @@ public class PendingAntiCompactionTest extends AbstractPendingAntiCompactionTest
                     public void onFailure(Throwable throwable)
                     {
                     }
-                });
+                }, MoreExecutors.directExecutor());
                 assertTrue(cdl.await(1, TimeUnit.MINUTES));
             }
         }

test/unit/org/apache/cassandra/hints/HintsServiceTest.java

@@ -25,6 +25,7 @@ import javax.annotation.Nullable;
 
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -129,7 +130,7 @@ public class HintsServiceTest
             {
                 HintsService.instance.resumeDispatch();
             }
-        });
+        }, MoreExecutors.directExecutor());
 
         Futures.allAsList(
                 noMessagesWhilePaused,

test/unit/org/apache/cassandra/net/MockMessagingSpy.java

@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
 import com.google.common.util.concurrent.AbstractFuture;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +54,7 @@ public class MockMessagingSpy
      */
     public ListenableFuture<Message<?>> captureMockedMessage()
     {
-        return Futures.transform(captureMockedMessageN(1), (List<Message<?>> result) -> result.isEmpty() ? null : result.get(0));
+        return Futures.transform(captureMockedMessageN(1), (List<Message<?>> result) -> result.isEmpty() ? null : result.get(0), MoreExecutors.directExecutor());
     }
 
     /**
@@ -89,7 +90,7 @@ public class MockMessagingSpy
      */
     public ListenableFuture<Message<?>> captureMessageOut()
     {
-        return Futures.transform(captureMessageOut(1), (List<Message<?>> result) -> result.isEmpty() ? null : result.get(0));
+        return Futures.transform(captureMessageOut(1), (List<Message<?>> result) -> result.isEmpty() ? null : result.get(0), MoreExecutors.directExecutor());
     }
 
     /**
@@ -231,4 +232,4 @@ public class MockMessagingSpy
             }
         }
     }
-}
+}

test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java

@@ -24,6 +24,8 @@ import java.util.concurrent.TimeUnit;
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.MoreExecutors;
+
 import org.apache.cassandra.locator.RangesAtEndpoint;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -130,7 +132,7 @@ public class StreamingTransferTest
             {
                 fail();
             }
-        });
+        }, MoreExecutors.directExecutor());
         // should be complete immediately
         futureResult.get(100, TimeUnit.MILLISECONDS);
     }

test/unit/org/apache/cassandra/tools/BulkLoaderTest.java

@@ -63,4 +63,76 @@ public class BulkLoaderTest extends ToolsTester
         assertKeyspaceNotLoaded();
         assertServerNotLoaded();
     }
+
+    @Test
+    public void testBulkLoader_WithArgs1() throws Exception
+    {
+        try
+        {
+            runTool(0, "org.apache.cassandra.tools.BulkLoader", "-d", "127.9.9.1", "--port", "9042", sstableDirName("legacy_sstables", "legacy_ma_simple"));
+            fail();
+        }
+        catch (RuntimeException e)
+        {
+            if (!(e.getCause() instanceof BulkLoadException))
+                throw e;
+            if (!(e.getCause().getCause() instanceof NoHostAvailableException))
+                throw e;
+        }
+        assertNoUnexpectedThreadsStarted(null, new String[]{"globalEventExecutor-1-1", "globalEventExecutor-1-2"});
+        assertSchemaNotLoaded();
+        assertCLSMNotLoaded();
+        assertSystemKSNotLoaded();
+        assertKeyspaceNotLoaded();
+        assertServerNotLoaded();
+    }
+
+    @Test
+    public void testBulkLoader_WithArgs2() throws Exception
+    {
+        try
+        {
+            runTool(0, "org.apache.cassandra.tools.BulkLoader", "-d", "127.9.9.1:9042", "--port", "9041", sstableDirName("legacy_sstables", "legacy_ma_simple"));
+            fail();
+        }
+        catch (RuntimeException e)
+        {
+            if (!(e.getCause() instanceof BulkLoadException))
+                throw e;
+            if (!(e.getCause().getCause() instanceof NoHostAvailableException))
+                throw e;
+        }
+        assertNoUnexpectedThreadsStarted(null, new String[]{"globalEventExecutor-1-1", "globalEventExecutor-1-2"});
+        assertSchemaNotLoaded();
+        assertCLSMNotLoaded();
+        assertSystemKSNotLoaded();
+        assertKeyspaceNotLoaded();
+        assertServerNotLoaded();
+    }
+
+    @Test(expected = NoHostAvailableException.class)
+    public void testBulkLoader_WithArgs3() throws Throwable
+    {
+        try
+        {
+            runTool(1, "org.apache.cassandra.tools.BulkLoader", "-d", "127.9.9.1", "--port", "9041", sstableDirName("legacy_sstables", "legacy_ma_simple"));
+        }
+        catch (RuntimeException e)
+        {
+            throw e.getCause().getCause();
+        }
+    }
+
+    @Test(expected = NoHostAvailableException.class)
+    public void testBulkLoader_WithArgs4() throws Throwable
+    {
+        try
+        {
+            runTool(1, "org.apache.cassandra.tools.BulkLoader", "-d", "127.9.9.1:9041", sstableDirName("legacy_sstables", "legacy_ma_simple"));
+        }
+        catch (RuntimeException e)
+        {
+            throw e.getCause().getCause();
+        }
+    }
 }

tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java

@@ -39,7 +39,6 @@ public class SettingsNode implements Serializable
     public final List<String> nodes;
     public final boolean isWhiteList;
     public final String datacenter;
-    public final boolean allowServerPortDiscovery;
 
     public SettingsNode(Options options)
     {
@@ -72,7 +71,6 @@ public class SettingsNode implements Serializable
 
         isWhiteList = options.whitelist.setByUser();
         datacenter = options.datacenter.value();
-        allowServerPortDiscovery = options.allowServerPortDiscovery.setByUser();
     }
 
     public Set<String> resolveAllPermitted(StressSettings settings)
@@ -145,13 +143,12 @@ public class SettingsNode implements Serializable
         final OptionSimple datacenter = new OptionSimple("datacenter=", ".*", null, "Datacenter used for DCAwareRoundRobinLoadPolicy", false);
         final OptionSimple whitelist = new OptionSimple("whitelist", "", null, "Limit communications to the provided nodes", false);
         final OptionSimple file = new OptionSimple("file=", ".*", null, "Node file (one per line)", false);
-        final OptionSimple allowServerPortDiscovery = new OptionSimple("allow_server_port_discovery", "", null, "Allow Java client to discover server client port numbers", false);
         final OptionSimple list = new OptionSimple("", "[^=,]+(,[^=,]+)*", "localhost", "comma delimited list of nodes", false);
 
         @Override
         public List<? extends Option> options()
         {
-            return Arrays.asList(datacenter, whitelist, file, allowServerPortDiscovery, list);
+            return Arrays.asList(datacenter, whitelist, file, list);
         }
     }
 
@@ -161,7 +158,6 @@ public class SettingsNode implements Serializable
         out.println("  Nodes: " + nodes);
         out.println("  Is White List: " + isWhiteList);
         out.println("  Datacenter: " + datacenter);
-        out.println("  Allow server port discovery: " + allowServerPortDiscovery);
     }
 
     public static SettingsNode get(Map<String, String[]> clArgs)

tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java

@@ -54,7 +54,6 @@ public class JavaDriverClient
     private Cluster cluster;
     private Session session;
     private final LoadBalancingPolicy loadBalancingPolicy;
-    private final boolean allowServerPortDiscovery;
 
     private static final ConcurrentMap<String, PreparedStatement> stmts = new ConcurrentHashMap<>();
 
@@ -74,7 +73,6 @@ public class JavaDriverClient
         this.encryptionOptions = encryptionOptions;
         this.loadBalancingPolicy = loadBalancingPolicy(settings);
         this.connectionsPerHost = settings.mode.connectionsPerHost == null ? 8 : settings.mode.connectionsPerHost;
-        this.allowServerPortDiscovery = settings.node.allowServerPortDiscovery;
 
         int maxThreadCount = 0;
         if (settings.rate.auto)
@@ -136,8 +134,6 @@ public class JavaDriverClient
                                                 .withoutJMXReporting()
                                                 .withProtocolVersion(protocolVersion)
                                                 .withoutMetrics(); // The driver uses metrics 3 with conflict with our version
-        if (allowServerPortDiscovery)
-            clusterBuilder = clusterBuilder.allowServerPortDiscovery();
 
         if (loadBalancingPolicy != null)
             clusterBuilder.withLoadBalancingPolicy(loadBalancingPolicy);