diff --git a/CHANGES.txt b/CHANGES.txt index 1ecda8bc9a..a72386aaed 100644 --- a/CHANGES.txt +++ b/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) diff --git a/build.xml b/build.xml index 3eb074fa22..73539df54a 100644 --- a/build.xml +++ b/build.xml @@ -487,7 +487,7 @@ - + @@ -557,15 +557,13 @@ - @@ -647,9 +645,7 @@ - @@ -666,9 +662,7 @@ artifactId="cassandra-parent" version="${version}"/> - @@ -743,14 +737,12 @@ - diff --git a/lib/cassandra-driver-core-3.4.0-SNAPSHOT-shaded.jar b/lib/cassandra-driver-core-3.4.0-SNAPSHOT-shaded.jar deleted file mode 100644 index 1290dc3e31..0000000000 Binary files a/lib/cassandra-driver-core-3.4.0-SNAPSHOT-shaded.jar and /dev/null differ diff --git a/lib/cassandra-driver-core-3.6.0-shaded.jar b/lib/cassandra-driver-core-3.6.0-shaded.jar new file mode 100644 index 0000000000..ea06b02455 Binary files /dev/null and b/lib/cassandra-driver-core-3.6.0-shaded.jar differ diff --git a/lib/guava-23.3-jre.jar b/lib/guava-23.3-jre.jar deleted file mode 100644 index b13e275fcf..0000000000 Binary files a/lib/guava-23.3-jre.jar and /dev/null differ diff --git a/lib/guava-27.0-jre.jar b/lib/guava-27.0-jre.jar new file mode 100644 index 0000000000..1bdd007424 Binary files /dev/null and b/lib/guava-27.0-jre.jar differ diff --git a/src/java/org/apache/cassandra/audit/AuditLogManager.java b/src/java/org/apache/cassandra/audit/AuditLogManager.java index d11eaa0ee7..bc2126ca7a 100644 --- a/src/java/org/apache/cassandra/audit/AuditLogManager.java +++ b/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; diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java index 4bbb861664..77ad95f29b 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java +++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java @@ -287,12 +287,10 @@ public class CqlBulkRecordWriter extends RecordWriter> public ExternalClient(Configuration conf) { super(resolveHostAddresses(conf), - CqlConfigHelper.getOutputNativePort(conf), ConfigHelper.getOutputInitialPort(conf), ConfigHelper.getOutputKeyspaceUserName(conf), ConfigHelper.getOutputKeyspacePassword(conf), - CqlConfigHelper.getSSLOptions(conf).orNull(), - CqlConfigHelper.getAllowServerPortDiscovery(conf)); + CqlConfigHelper.getSSLOptions(conf).orNull()); } private static Collection resolveHostAddresses(Configuration conf) diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java index 3a47a728b2..f9a6f3ae0f 100644 --- a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java +++ b/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")); - } - } diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java index 60fc3ba314..336e2eae00 100644 --- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java +++ b/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); } }); diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableFormat.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableFormat.java index 38a7f5745a..14f660258f 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableFormat.java +++ b/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; diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java index f682bfb164..3740070d13 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -391,7 +391,7 @@ public class RepairJob extends AbstractFuture 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 implements Runnable // failure is handled at root of job chain public void onFailure(Throwable t) {} - }); + }, MoreExecutors.directExecutor()); currentTask = nextTask; } // start running tasks diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java index 7e931e81ca..f1f12ddc12 100644 --- a/src/java/org/apache/cassandra/repair/RepairRunnable.java +++ b/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> 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); diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index 40f3dbe04a..3483e593ae 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -344,7 +344,7 @@ public class RepairSession extends AbstractFuture implement Tracing.traceRepair("Session completed with the following error: {}", t); forceShutdown(t); } - }); + }, MoreExecutors.directExecutor()); } public void terminate() diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java index 8f1759afd1..39549bd903 100644 --- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java +++ b/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; } diff --git a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java b/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java index 935bba8dfc..6475794e0c 100644 --- a/src/java/org/apache/cassandra/repair/consistent/LocalSessions.java +++ b/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) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a7c125c865..777e40e5d0 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/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()); } /** diff --git a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java index 4de63be907..3268ecf8ff 100644 --- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java +++ b/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 public void addEventListener(StreamEventHandler listener) { - Futures.addCallback(this, listener); + Futures.addCallback(this, listener, MoreExecutors.directExecutor()); eventListeners.add(listener); } diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java index 2ca2a3d5b2..42c2bf37ff 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoader.java +++ b/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), - options.allowServerPortDiscovery), + buildSSLOptions(options.clientEncOptions)), handler, options.connectionsPerHost, options.targetKeyspace); @@ -275,15 +273,13 @@ public class BulkLoader private final EncryptionOptions.ServerEncryptionOptions serverEncOptions; public ExternalClient(Set hosts, - int nativePort, int storagePort, AuthProvider authProvider, int sslStoragePort, EncryptionOptions.ServerEncryptionOptions serverEncryptionOptions, - SSLOptions sslOptions, - boolean allowServerPortDiscovery) + SSLOptions sslOptions) { - super(hosts, nativePort, storagePort, authProvider, sslOptions, allowServerPortDiscovery); + super(hosts, storagePort, authProvider, sslOptions); this.sslStoragePort = sslStoragePort; serverEncOptions = serverEncryptionOptions; } diff --git a/src/java/org/apache/cassandra/tools/LoaderOptions.java b/src/java/org/apache/cassandra/tools/LoaderOptions.java index 7ad3299e29..9acf8a5fbf 100644 --- a/src/java/org/apache/cassandra/tools/LoaderOptions.java +++ b/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 hosts; public final Set 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 ignoresArg = new HashSet<>(); Set hosts = new HashSet<>(); Set 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; } diff --git a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java index bb0ee2559a..9763d7e8b6 100644 --- a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java +++ b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java @@ -46,37 +46,28 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client { protected final Map tables; private final Collection hosts; - private final int port; private final int storagePort; private final AuthProvider authProvider; private final SSLOptions sslOptions; - private final boolean allowServerPortDiscovery; - - public NativeSSTableLoaderClient(Collection hosts, int nativePort, int storagePort, String username, String password, SSLOptions sslOptions, boolean allowServerPortDiscovery) + public NativeSSTableLoaderClient(Collection hosts, int storagePort, String username, String password, SSLOptions sslOptions) { - this(hosts, nativePort, storagePort, new PlainTextAuthProvider(username, password), sslOptions, allowServerPortDiscovery); + this(hosts, storagePort, new PlainTextAuthProvider(username, password), sslOptions); } - public NativeSSTableLoaderClient(Collection hosts, int nativePort, int storagePort, AuthProvider authProvider, SSLOptions sslOptions, boolean allowServerPortDiscovery) + public NativeSSTableLoaderClient(Collection 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 hostAddresses = hosts.stream().map(host -> host.getAddress()).collect(Collectors.toSet()); - Cluster.Builder builder = Cluster.builder().addContactPoints(hostAddresses).withPort(port).allowBetaProtocolVersion(); - - if (allowServerPortDiscovery) - builder = builder.allowServerPortDiscovery(); + Cluster.Builder builder = Cluster.builder().addContactPointsWithPorts(hosts).allowBetaProtocolVersion(); 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; - if (allowServerPortDiscovery) - { - portToUse = endpoint.getBroadcastAddressOptPort().portOrElse(storagePort); - } - else - { - portToUse = storagePort; - } + int broadcastPort = endpoint.getBroadcastSocketAddress().getPort(); + // use port from broadcast address if set. + int portToUse = broadcastPort != 0 ? broadcastPort : storagePort; addRangeForEndpoint(range, InetAddressAndPort.getByNameOverrideDefaults(endpoint.getAddress().getHostAddress(), portToUse)); } } diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index 9c4f22e51e..8f3a52a0bf 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/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(); diff --git a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java b/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java index b140813f48..6bef001050 100644 --- a/test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java +++ b/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)); } } diff --git a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java b/test/unit/org/apache/cassandra/hints/HintsServiceTest.java index 77783311c7..dddf336ed5 100644 --- a/test/unit/org/apache/cassandra/hints/HintsServiceTest.java +++ b/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, diff --git a/test/unit/org/apache/cassandra/net/MockMessagingSpy.java b/test/unit/org/apache/cassandra/net/MockMessagingSpy.java index bf4c2267c8..c61c3013f5 100644 --- a/test/unit/org/apache/cassandra/net/MockMessagingSpy.java +++ b/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> captureMockedMessage() { - return Futures.transform(captureMockedMessageN(1), (List> result) -> result.isEmpty() ? null : result.get(0)); + return Futures.transform(captureMockedMessageN(1), (List> result) -> result.isEmpty() ? null : result.get(0), MoreExecutors.directExecutor()); } /** @@ -89,7 +90,7 @@ public class MockMessagingSpy */ public ListenableFuture> captureMessageOut() { - return Futures.transform(captureMessageOut(1), (List> result) -> result.isEmpty() ? null : result.get(0)); + return Futures.transform(captureMessageOut(1), (List> result) -> result.isEmpty() ? null : result.get(0), MoreExecutors.directExecutor()); } /** @@ -231,4 +232,4 @@ public class MockMessagingSpy } } } -} +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java b/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java index 909e221ae2..d88f379fec 100644 --- a/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java +++ b/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); } diff --git a/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java b/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java index 104f288d84..6ed38a0893 100644 --- a/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java +++ b/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(); + } + } } diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java index 24c10bf8d8..8a484c72a9 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsNode.java @@ -39,7 +39,6 @@ public class SettingsNode implements Serializable public final List 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 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 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 clArgs) diff --git a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java index 36361f7de2..72487dc8e0 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java +++ b/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 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);