diff --git a/CHANGES.txt b/CHANGES.txt index 255b46d7aa..36beb3c27f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -37,6 +37,7 @@ * Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495) * Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544) Merged from 4.1: + * Fix BulkLoader to load entireSSTableThrottle and entireSSTableInterDcThrottle (CASSANDRA-17677) * Fix a race condition where a keyspace can be oopened while it is being removed (CASSANDRA-17658) * DatabaseDescriptor will set the default failure detector during client initialization (CASSANDRA-17782) * Avoid initializing schema via SystemKeyspace.getPreferredIP() with the BulkLoader tool (CASSANDRA-17740) diff --git a/doc/modules/cassandra/pages/operating/bulk_loading.adoc b/doc/modules/cassandra/pages/operating/bulk_loading.adoc index cf65b12ede..0d85ae4d1a 100644 --- a/doc/modules/cassandra/pages/operating/bulk_loading.adoc +++ b/doc/modules/cassandra/pages/operating/bulk_loading.adoc @@ -80,13 +80,23 @@ The following options are supported, with `-d,--nodes ` required: -d,--nodes Required. Try to connect to these hosts (comma separated) initially for ring information +--entire-sstable-throttle-mib Entire SSTable throttle + speed in MiB/s (default 0 for unlimited). + +--entire-sstable-inter-dc-throttle-mib + Entire SSTable inter-datacenter throttle + speed in MiB/s (default 0 for unlimited). + -f,--conf-path cassandra.yaml file path for streaming throughput and client/server SSL. -h,--help Display this help message -i,--ignore Don't stream to this (comma separated) list of nodes --idct,--inter-dc-throttle Inter-datacenter throttle speed in Mbits (default unlimited) +-idct,--inter-dc-throttle (deprecated) Inter-datacenter throttle speed in Mbits (default 0 for unlimited). + Use --inter-dc-throttle-mib instead. + +--inter-dc-throttle-mib Inter-datacenter throttle speed in MiB/s (default 0 for unlimited) -k,--target-keyspace Target keyspace name @@ -111,8 +121,10 @@ The following options are supported, with `-d,--nodes ` required: for TLS internode communication (default 7001) -st,--store-type Client SSL: type of store --t,--throttle Throttle - speed in Mbits (default unlimited) +-t,--throttle (deprecated) Throttle speed in Mbits (default 0 for unlimited). + Use --throttle-mib instead. +--throttle-mib Throttle + speed in MiB/s (default 0 for unlimited) -ts,--truststore Client SSL: full path to truststore -tspw,--truststore-password Client SSL: diff --git a/doc/modules/cassandra/pages/tools/sstable/sstableloader.adoc b/doc/modules/cassandra/pages/tools/sstable/sstableloader.adoc index c03bd75e2e..e980dc2e00 100644 --- a/doc/modules/cassandra/pages/tools/sstable/sstableloader.adoc +++ b/doc/modules/cassandra/pages/tools/sstable/sstableloader.adoc @@ -45,10 +45,23 @@ internode communication (default 7001) |--no-progress |don't display progress -|-t, --throttle |throttle speed in Mbits (default unlimited) +|-t, --throttle |(deprecated) throttle speed in Mbits +(default 0 for unlimited) Use --throttle-mib instead -|-idct, --inter-dc-throttle |inter-datacenter -throttle speed in Mbits (default unlimited) +|--throttle-mib |throttle speed in MiB/s +(default 0 for unlimited) + +|-idct, --inter-dc-throttle |(deprecated) inter-datacenter +throttle speed in Mbits (default 0 for unlimited) Use --inter-dc-throttle-mib instead + +|--inter-dc-throttle-mib |inter-datacenter +throttle speed in MiB/s (default 0 for unlimited) + +|--entire-sstable-throttle-mib |entire SSTable throttle +speed in MiB/s (default 0 for unlimited) + +|--entire-sstable-inter-dc-throttle-mib |entire +SSTable inter-datacenter throttle speed in MiB/s (default 0 for unlimited) |-cph, --connections-per-host |number of concurrent connections-per-host diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 001a7a428c..42b5e27fba 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2120,6 +2120,11 @@ public class DatabaseDescriptor conf.stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND); } + public static void setStreamThroughputOutboundBytesPerSec(long value) + { + conf.stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, BYTES_PER_SECOND); + } + public static void setStreamThroughputOutboundMegabitsPerSec(int value) { conf.stream_throughput_outbound = DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(value); @@ -2180,6 +2185,11 @@ public class DatabaseDescriptor conf.inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND); } + public static void setInterDCStreamThroughputOutboundBytesPerSec(long value) + { + conf.inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, BYTES_PER_SECOND); + } + public static void setInterDCStreamThroughputOutboundMegabitsPerSec(int value) { conf.inter_dc_stream_throughput_outbound = DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(value); diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java index a3a296b97a..8d5a1d4c67 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoader.java +++ b/src/java/org/apache/cassandra/tools/BulkLoader.java @@ -23,20 +23,26 @@ import java.util.Set; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; -import com.datastax.driver.core.AuthProvider; -import com.datastax.driver.core.RemoteEndpointAwareJdkSSLOptions; -import com.datastax.driver.core.SSLOptions; import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; import org.apache.commons.cli.Option; import org.apache.commons.cli.Options; +import com.datastax.driver.core.AuthProvider; +import com.datastax.driver.core.RemoteEndpointAwareJdkSSLOptions; +import com.datastax.driver.core.SSLOptions; import com.datastax.shaded.netty.channel.socket.SocketChannel; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.EncryptionOptions; import org.apache.cassandra.io.sstable.SSTableLoader; import org.apache.cassandra.security.SSLFactory; -import org.apache.cassandra.streaming.*; +import org.apache.cassandra.streaming.ProgressInfo; +import org.apache.cassandra.streaming.SessionInfo; +import org.apache.cassandra.streaming.StreamEvent; +import org.apache.cassandra.streaming.StreamEventHandler; +import org.apache.cassandra.streaming.StreamResultFuture; +import org.apache.cassandra.streaming.StreamState; +import org.apache.cassandra.streaming.StreamingChannel; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.NativeSSTableLoaderClient; @@ -46,7 +52,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; public class BulkLoader { - public static void main(String args[]) throws BulkLoadException + public static void main(String[] args) throws BulkLoadException { LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); load(options); @@ -68,9 +74,11 @@ public class BulkLoader options.connectionsPerHost, options.targetKeyspace, options.targetTable); - DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle); - DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(options.interDcThrottle); - StreamResultFuture future = null; + DatabaseDescriptor.setStreamThroughputOutboundBytesPerSec(options.throttleBytes); + DatabaseDescriptor.setInterDCStreamThroughputOutboundBytesPerSec(options.interDcThrottleBytes); + DatabaseDescriptor.setEntireSSTableStreamThroughputOutboundMebibytesPerSec(options.entireSSTableThrottleMebibytes); + DatabaseDescriptor.setEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec(options.entireSSTableInterDcThrottleMebibytes); + StreamResultFuture future; ProgressIndicator indicator = new ProgressIndicator(); try @@ -120,7 +128,7 @@ public class BulkLoader // Return true when everything is at 100% static class ProgressIndicator implements StreamEventHandler { - private long start; + private final long start; private long lastProgress; private long lastTime; @@ -219,7 +227,7 @@ public class BulkLoader } sb.append(" (avg: ").append(FBUtilities.prettyPrintMemoryPerSecond(totalProgress, time - start)).append(")"); - System.out.println(sb.toString()); + System.out.println(sb); } } @@ -241,7 +249,7 @@ public class BulkLoader sb.append(String.format(" %-24s: %-10s%n", "Total duration ", durationMS + " ms")); sb.append(String.format(" %-24s: %-10s%n", "Average transfer rate ", FBUtilities.prettyPrintMemoryPerSecond(lastProgress, end - start))); sb.append(String.format(" %-24s: %-10s%n", "Peak transfer rate ", FBUtilities.prettyPrintMemoryPerSecond(peak))); - System.out.println(sb.toString()); + System.out.println(sb); } } diff --git a/src/java/org/apache/cassandra/tools/LoaderOptions.java b/src/java/org/apache/cassandra/tools/LoaderOptions.java index 57d9f1e251..03768af2e4 100644 --- a/src/java/org/apache/cassandra/tools/LoaderOptions.java +++ b/src/java/org/apache/cassandra/tools/LoaderOptions.java @@ -22,25 +22,37 @@ package org.apache.cassandra.tools; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; -import java.net.*; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.MalformedURLException; +import java.net.UnknownHostException; import java.util.HashSet; import java.util.Set; import com.google.common.base.Throwables; import com.google.common.net.HostAndPort; +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.GnuParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import org.apache.cassandra.config.*; +import com.datastax.driver.core.AuthProvider; +import com.datastax.driver.core.PlainTextAuthProvider; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DataRateSpec; +import org.apache.cassandra.config.EncryptionOptions; +import org.apache.cassandra.config.YamlConfigurationLoader; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.util.File; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.tools.BulkLoader.CmdLineOptions; -import com.datastax.driver.core.AuthProvider; -import com.datastax.driver.core.PlainTextAuthProvider; -import org.apache.commons.cli.*; -import org.apache.commons.lang3.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND; public class LoaderOptions { @@ -60,10 +72,25 @@ public class LoaderOptions public static final String IGNORE_NODES_OPTION = "ignore"; public static final String CONNECTIONS_PER_HOST = "connections-per-host"; public static final String CONFIG_PATH = "conf-path"; + + /** + * Throttle defined in megabits per second. CASSANDRA-10637 introduced a builder and is the preferred way to + * provide options instead of using these constant fields. + * @deprecated Use {@code throttle-mib} instead + */ + @Deprecated public static final String THROTTLE_MBITS = "throttle"; + public static final String THROTTLE_MEBIBYTES = "throttle-mib"; + /** + * Inter-datacenter throttle defined in megabits per second. CASSANDRA-10637 introduced a builder and is the + * preferred way to provide options instead of using these constant fields. + * @deprecated Use {@code inter-dc-throttle-mib} instead + */ + @Deprecated public static final String INTER_DC_THROTTLE_MBITS = "inter-dc-throttle"; - public static final String ENTIRE_SSTABLE_THROTTLE_MBITS = "entire-sstable-throttle"; - public static final String ENTIRE_SSTABLE_INTER_DC_THROTTLE_MBITS = "entire-sstable-inter-dc-throttle"; + public static final String INTER_DC_THROTTLE_MEBIBYTES = "inter-dc-throttle-mib"; + public static final String ENTIRE_SSTABLE_THROTTLE_MEBIBYTES = "entire-sstable-throttle-mib"; + public static final String ENTIRE_SSTABLE_INTER_DC_THROTTLE_MEBIBYTES = "entire-sstable-inter-dc-throttle-mib"; public static final String TOOL_NAME = "sstableloader"; public static final String TARGET_KEYSPACE = "target-keyspace"; public static final String TARGET_TABLE = "target-table"; @@ -86,10 +113,10 @@ public class LoaderOptions public final String user; public final String passwd; public final AuthProvider authProvider; - public final int throttle; - public final int interDcThrottle; - public final int entireSSTableThrottle; - public final int entireSSTableInterDcThrottle; + public final long throttleBytes; + public final long interDcThrottleBytes; + public final int entireSSTableThrottleMebibytes; + public final int entireSSTableInterDcThrottleMebibytes; public final int storagePort; public final int sslStoragePort; public final EncryptionOptions clientEncOptions; @@ -110,10 +137,10 @@ public class LoaderOptions user = builder.user; passwd = builder.passwd; authProvider = builder.authProvider; - throttle = builder.throttle; - interDcThrottle = builder.interDcThrottle; - entireSSTableThrottle = builder.entireSSTableThrottle; - entireSSTableInterDcThrottle = builder.entireSSTableInterDcThrottle; + throttleBytes = builder.throttleBytes; + interDcThrottleBytes = builder.interDcThrottleBytes; + entireSSTableThrottleMebibytes = builder.entireSSTableThrottleMebibytes; + entireSSTableInterDcThrottleMebibytes = builder.entireSSTableInterDcThrottleMebibytes; storagePort = builder.storagePort; sslStoragePort = builder.sslStoragePort; clientEncOptions = builder.clientEncOptions; @@ -136,10 +163,11 @@ public class LoaderOptions String passwd; String authProviderName; AuthProvider authProvider; - int throttle = 0; - int interDcThrottle = 0; - int entireSSTableThrottle = 0; - int entireSSTableInterDcThrottle = 0; + long throttleBytes = 0; + long interDcThrottleBytes = 0; + int entireSSTableThrottleMebibytes = 0; + int entireSSTableInterDcThrottleMebibytes = 0; + int storagePort; int sslStoragePort; EncryptionOptions clientEncOptions = new EncryptionOptions(); @@ -228,27 +256,60 @@ public class LoaderOptions return this; } - public Builder throttle(int throttle) + public Builder throttleMebibytes(int throttleMebibytes) { - this.throttle = throttle; + this.throttleBytes = (long) MEBIBYTES_PER_SECOND.toBytesPerSecond(throttleMebibytes); return this; } + @Deprecated + public Builder throttle(int throttleMegabits) + { + this.throttleBytes = (long) DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(throttleMegabits).toBytesPerSecond(); + return this; + } + + public Builder interDcThrottleMebibytes(int interDcThrottleMebibytes) + { + this.interDcThrottleBytes = (long) MEBIBYTES_PER_SECOND.toBytesPerSecond(interDcThrottleMebibytes); + return this; + } + + public Builder interDcThrottleMegabits(int interDcThrottleMegabits) + { + this.interDcThrottleBytes = (long) DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(interDcThrottleMegabits).toBytesPerSecond(); + return this; + } + + @Deprecated public Builder interDcThrottle(int interDcThrottle) { - this.interDcThrottle = interDcThrottle; + return interDcThrottleMegabits(interDcThrottle); + } + + public Builder entireSSTableThrottleMebibytes(int entireSSTableThrottleMebibytes) + { + this.entireSSTableThrottleMebibytes = entireSSTableThrottleMebibytes; return this; } + @Deprecated public Builder entireSSTableThrottle(int entireSSTableThrottle) { - this.entireSSTableThrottle = entireSSTableThrottle; + this.entireSSTableThrottleMebibytes = entireSSTableThrottle; return this; } + public Builder entireSSTableInterDcThrottleMebibytes(int entireSSTableInterDcThrottleMebibytes) + { + this.entireSSTableInterDcThrottleMebibytes = entireSSTableInterDcThrottleMebibytes; + return this; + } + + @Deprecated public Builder entireSSTableInterDcThrottle(int entireSSTableInterDcThrottle) { - this.entireSSTableInterDcThrottle = entireSSTableInterDcThrottle; + this.entireSSTableInterDcThrottleMebibytes = entireSSTableInterDcThrottle; return this; } @@ -475,7 +536,7 @@ public class LoaderOptions connectionsPerHost = Integer.parseInt(cmd.getOptionValue(CONNECTIONS_PER_HOST)); } - throttle = config.stream_throughput_outbound.toMebibytesPerSecondAsInt(); + throttleBytes = config.stream_throughput_outbound.toBytesPerSecondAsInt(); if (cmd.hasOption(SSL_STORAGE_PORT_OPTION)) logger.info("ssl storage port is deprecated and not used, all communication goes though storage port " + @@ -520,28 +581,48 @@ public class LoaderOptions System.exit(1); } + if (cmd.hasOption(THROTTLE_MBITS) && cmd.hasOption(THROTTLE_MEBIBYTES)) + { + errorMsg(String.format("Both '%s' and '%s' were provided. Please only provide one of the two options", THROTTLE_MBITS, THROTTLE_MEBIBYTES), options); + } + + if (cmd.hasOption(INTER_DC_THROTTLE_MBITS) && cmd.hasOption(INTER_DC_THROTTLE_MEBIBYTES)) + { + errorMsg(String.format("Both '%s' and '%s' were provided. Please only provide one of the two options", INTER_DC_THROTTLE_MBITS, INTER_DC_THROTTLE_MEBIBYTES), options); + } + if (cmd.hasOption(THROTTLE_MBITS)) { - throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS)); + throttle(Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS))); + } + + if (cmd.hasOption(THROTTLE_MEBIBYTES)) + { + throttleMebibytes(Integer.parseInt(cmd.getOptionValue(THROTTLE_MEBIBYTES))); } if (cmd.hasOption(INTER_DC_THROTTLE_MBITS)) { - interDcThrottle = Integer.parseInt(cmd.getOptionValue(INTER_DC_THROTTLE_MBITS)); + interDcThrottleMegabits(Integer.parseInt(cmd.getOptionValue(INTER_DC_THROTTLE_MBITS))); } - if (cmd.hasOption(ENTIRE_SSTABLE_THROTTLE_MBITS)) + if (cmd.hasOption(INTER_DC_THROTTLE_MEBIBYTES)) { - entireSSTableThrottle = Integer.parseInt(cmd.getOptionValue(ENTIRE_SSTABLE_THROTTLE_MBITS)); + interDcThrottleMebibytes(Integer.parseInt(cmd.getOptionValue(INTER_DC_THROTTLE_MEBIBYTES))); } - if (cmd.hasOption(ENTIRE_SSTABLE_INTER_DC_THROTTLE_MBITS)) + if (cmd.hasOption(ENTIRE_SSTABLE_THROTTLE_MEBIBYTES)) { - entireSSTableInterDcThrottle = Integer.parseInt(cmd.getOptionValue(ENTIRE_SSTABLE_INTER_DC_THROTTLE_MBITS)); + entireSSTableThrottleMebibytes(Integer.parseInt(cmd.getOptionValue(ENTIRE_SSTABLE_THROTTLE_MEBIBYTES))); + } + + if (cmd.hasOption(ENTIRE_SSTABLE_INTER_DC_THROTTLE_MEBIBYTES)) + { + entireSSTableInterDcThrottleMebibytes(Integer.parseInt(cmd.getOptionValue(ENTIRE_SSTABLE_INTER_DC_THROTTLE_MEBIBYTES))); } if (cmd.hasOption(SSL_TRUSTSTORE) || cmd.hasOption(SSL_TRUSTSTORE_PW) || - cmd.hasOption(SSL_KEYSTORE) || cmd.hasOption(SSL_KEYSTORE_PW)) + cmd.hasOption(SSL_KEYSTORE) || cmd.hasOption(SSL_KEYSTORE_PW)) { clientEncOptions = clientEncOptions.withEnabled(true); } @@ -691,10 +772,12 @@ public class LoaderOptions options.addOption("p", NATIVE_PORT_OPTION, "native transport port", "port used for native connection (default 9042)"); options.addOption("sp", STORAGE_PORT_OPTION, "storage port", "port used for internode communication (default 7000)"); options.addOption("ssp", SSL_STORAGE_PORT_OPTION, "ssl storage port", "port used for TLS internode communication (default 7001), this option is deprecated, all communication goes through storage port which handles encrypted communication as well"); - options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)"); - options.addOption("idct", INTER_DC_THROTTLE_MBITS, "inter-dc-throttle", "inter-datacenter throttle speed in Mbits (default unlimited)"); - options.addOption("e", ENTIRE_SSTABLE_THROTTLE_MBITS, "entire-sstable-throttle", "entire SSTable throttle speed in Mbits (default unlimited)"); - options.addOption("eidct", ENTIRE_SSTABLE_INTER_DC_THROTTLE_MBITS, "entire-sstable-inter-dc-throttle", "entire SSTable inter-datacenter throttle speed in Mbits (default unlimited)"); + options.addOption("t", THROTTLE_MBITS, "throttle", "throttle speed in Mbps (default 0 for unlimited), this option is deprecated, use \"throttle-mib\" instead"); + options.addOption(null, THROTTLE_MEBIBYTES, "throttle-mib", "throttle speed in MiB/s (default 0 for unlimited)"); + options.addOption("idct", INTER_DC_THROTTLE_MBITS, "inter-dc-throttle", "inter-datacenter throttle speed in Mbps (default 0 for unlimited), this option is deprecated, use \"inter-dc-throttle-mib\" instead"); + options.addOption(null, INTER_DC_THROTTLE_MEBIBYTES, "inter-dc-throttle-mib", "inter-datacenter throttle speed in MiB/s (default 0 for unlimited)"); + options.addOption(null, ENTIRE_SSTABLE_THROTTLE_MEBIBYTES, "entire-sstable-throttle-mib", "entire SSTable throttle speed in MiB/s (default 0 for unlimited)"); + options.addOption(null, ENTIRE_SSTABLE_INTER_DC_THROTTLE_MEBIBYTES, "entire-sstable-inter-dc-throttle-mib", "entire SSTable inter-datacenter throttle speed in MiB/s (default 0 for unlimited)"); options.addOption("u", USER_OPTION, "username", "username for cassandra authentication"); options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication"); options.addOption("ap", AUTH_PROVIDER_OPTION, "auth provider", "custom AuthProvider class name for cassandra authentication"); diff --git a/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java b/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java index 2d9acf7cef..4b2090ffff 100644 --- a/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java +++ b/test/unit/org/apache/cassandra/tools/BulkLoaderTest.java @@ -21,6 +21,7 @@ package org.apache.cassandra.tools; import org.junit.Test; import com.datastax.driver.core.exceptions.NoHostAvailableException; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.tools.ToolRunner.ToolResult; import org.hamcrest.CoreMatchers; @@ -157,4 +158,50 @@ public class BulkLoaderTest extends OfflineToolUtils assertEquals(-1, tool.getExitCode()); throw tool.getException().getCause().getCause().getCause(); } + + @Test(expected = NoHostAvailableException.class) + public void testBulkLoader_WithArgs5() throws Throwable + { + ToolResult tool = ToolRunner.invokeClass(BulkLoader.class, + "-d", + "127.9.9.1:9041", + "--throttle", + "10", + "--inter-dc-throttle", + "15", + "--entire-sstable-throttle-mib", + "20", + "--entire-sstable-inter-dc-throttle-mib", + "25", + OfflineToolUtils.sstableDirName("legacy_sstables", "legacy_ma_simple")); + assertEquals(-1, tool.getExitCode()); + assertEquals(10 * 125_000, DatabaseDescriptor.getStreamThroughputOutboundBytesPerSec(), 0.0); + assertEquals(15 * 125_000, DatabaseDescriptor.getInterDCStreamThroughputOutboundBytesPerSec(), 0.0); + assertEquals(20, DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec(), 0.0); + assertEquals(25, DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec(), 0.0); + throw tool.getException().getCause().getCause().getCause(); + } + + @Test(expected = NoHostAvailableException.class) + public void testBulkLoader_WithArgs6() throws Throwable + { + ToolResult tool = ToolRunner.invokeClass(BulkLoader.class, + "-d", + "127.9.9.1:9041", + "--throttle-mib", + "3", + "--inter-dc-throttle-mib", + "4", + "--entire-sstable-throttle-mib", + "5", + "--entire-sstable-inter-dc-throttle-mib", + "6", + OfflineToolUtils.sstableDirName("legacy_sstables", "legacy_ma_simple")); + assertEquals(-1, tool.getExitCode()); + assertEquals(3 * 1024 * 1024, DatabaseDescriptor.getStreamThroughputOutboundBytesPerSec(), 0.0); + assertEquals(4 * 1024 * 1024, DatabaseDescriptor.getInterDCStreamThroughputOutboundBytesPerSec(), 0.0); + assertEquals(5, DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec(), 0.0); + assertEquals(6, DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec(), 0.0); + throw tool.getException().getCause().getCause().getCause(); + } } diff --git a/test/unit/org/apache/cassandra/tools/LoaderOptionsTest.java b/test/unit/org/apache/cassandra/tools/LoaderOptionsTest.java index 8efe2cd915..4fc2325677 100644 --- a/test/unit/org/apache/cassandra/tools/LoaderOptionsTest.java +++ b/test/unit/org/apache/cassandra/tools/LoaderOptionsTest.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.file.Paths; +import java.security.Permission; import com.google.common.net.HostAndPort; import org.junit.Test; @@ -30,24 +31,24 @@ import org.apache.cassandra.io.util.File; import static org.apache.cassandra.tools.OfflineToolUtils.sstableDirName; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; // LoaderOptionsTester for custom configuration public class LoaderOptionsTest { @Test - public void testNativePort() throws Exception { + public void testNativePort() throws Exception + { //Default Cassandra config File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); - String[] args = { "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple")}; + String[] args = { "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); assertEquals(9042, options.nativePort); // SSL Enabled Cassandra config config = new File(Paths.get(".", "test", "conf", "unit-test-conf/test-native-port.yaml").normalize()); - String[] args2 = { "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple")}; + String[] args2 = { "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; options = LoaderOptions.builder().parseArgs(args2).build(); assertEquals(9142, options.nativePort); @@ -58,7 +59,7 @@ public class LoaderOptionsTest // test native port set from command line config = new File(Paths.get(".", "test", "conf", "unit-test-conf/test-native-port.yaml").normalize().toFile()); - String[] args3 = {"-d", "127.9.9.1", "-p", "9300", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple")}; + String[] args3 = { "-d", "127.9.9.1", "-p", "9300", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; options = LoaderOptions.builder().parseArgs(args3).build(); assertEquals(9300, options.nativePort); @@ -77,32 +78,99 @@ public class LoaderOptionsTest public void testEncryptionSettings() throws Exception { String[] args = { "-d", "127.9.9.1", "-ts", "test.jks", "-tspw", "truststorePass1", "-ks", "test.jks", "-kspw", - "testdata1", "--ssl-ciphers", "TLS_RSA_WITH_AES_256_CBC_SHA", - "--ssl-alg", "SunX509", "--store-type", "JKS", "--ssl-protocol", "TLS", - sstableDirName("legacy_sstables", "legacy_ma_simple") }; + "testdata1", "--ssl-ciphers", "TLS_RSA_WITH_AES_256_CBC_SHA", + "--ssl-alg", "SunX509", "--store-type", "JKS", "--ssl-protocol", "TLS", + sstableDirName("legacy_sstables", "legacy_ma_simple") }; LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); assertEquals("test.jks", options.clientEncOptions.keystore); } + @Test + public void testThrottleDefaultSettings() + { + LoaderOptions options = LoaderOptions.builder().build(); + assertEquals(0, options.throttleBytes, 0); + assertEquals(0, options.interDcThrottleBytes, 0); + } + + @Test + public void testDeprecatedThrottleSettings() throws IOException + { + // Default Cassandra config + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "-t", "200", "-idct", "400", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); + // converts from megabits to bytes + assertEquals(200 * 125_000, options.throttleBytes, 0); + assertEquals(400 * 125_000, options.interDcThrottleBytes, 0); + } + + @Test + public void testDeprecatedThrottleSettingsWithLongSettingNames() throws IOException + { + // Default Cassandra config + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "--throttle", "200", "--inter-dc-throttle", "400", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); + // converts from megabits to bytes + assertEquals(200 * 125_000, options.throttleBytes, 0); + assertEquals(400 * 125_000, options.interDcThrottleBytes, 0); + } + + @Test + public void testThrottleSettingsWithLongSettingNames() throws IOException + { + // Default Cassandra config + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "--throttle-mib", "24", "--inter-dc-throttle-mib", "48", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); + // converts from mebibytes to bytes + assertEquals(24 * 1024 * 1024, options.throttleBytes, 0); + assertEquals(48 * 1024 * 1024, options.interDcThrottleBytes, 0); + } + + @Test + public void failsWhenThrottleSettingAndDeprecatedAreProvided() throws IOException + { + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "-t", "200", "-tmib", "200", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + + failureHelper(args, 1); + } + + @Test + public void failsWhenThrottleSettingAndDeprecatedAreProvidedWithLongSettingNames() throws IOException + { + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "--throttle", "200", "--throttle-mib", "200", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + + failureHelper(args, 1); + } + + @Test + public void failsWhenInterDCThrottleSettingAndDeprecatedAreProvided() throws IOException + { + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "-idct", "200", "-idctmib", "200", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + + failureHelper(args, 1); + } + + @Test + public void failsWhenInterDCThrottleSettingAndDeprecatedAreProvidedWithLongSettingNames() throws IOException + { + File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); + String[] args = { "--inter-dc-throttle", "200", "--inter-dc-throttle-mib", "200", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + + failureHelper(args, 1); + } + @Test public void testEntireSSTableDefaultSettings() { LoaderOptions options = LoaderOptions.builder().build(); - assertEquals(0, options.entireSSTableThrottle); - assertEquals(0, options.entireSSTableInterDcThrottle); - } - - @Test - public void testEntireSSTableSettings() throws IOException - { - // Default Cassandra config - File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); - String[] args = { "-e", "350", "-eidct", "600", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; - LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); - assertNotNull(options.entireSSTableThrottle); - assertEquals(350, options.entireSSTableThrottle); - assertNotNull(options.entireSSTableInterDcThrottle); - assertEquals(600, options.entireSSTableInterDcThrottle); + assertEquals(0, options.entireSSTableThrottleMebibytes); + assertEquals(0, options.entireSSTableInterDcThrottleMebibytes); } @Test @@ -110,12 +178,42 @@ public class LoaderOptionsTest { // Use long names for the args, i.e. entire-sstable-throttle File config = new File(Paths.get(".", "test", "conf", "cassandra.yaml").normalize()); - String[] args = new String[]{ "--entire-sstable-throttle", "350", "--entire-sstable-inter-dc-throttle", "600", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; + String[] args = new String[]{ "--entire-sstable-throttle-mib", "350", "--entire-sstable-inter-dc-throttle-mib", "600", "-d", "127.9.9.1", "-f", config.absolutePath(), sstableDirName("legacy_sstables", "legacy_ma_simple") }; LoaderOptions options = LoaderOptions.builder().parseArgs(args).build(); - assertNotNull(options.entireSSTableThrottle); - assertEquals(350, options.entireSSTableThrottle); - assertNotNull(options.entireSSTableInterDcThrottle); - assertEquals(600, options.entireSSTableInterDcThrottle); + assertEquals(350, options.entireSSTableThrottleMebibytes); + assertEquals(600, options.entireSSTableInterDcThrottleMebibytes); + } + + private void failureHelper(String[] args, int expectedErrorCode) + { + // install security manager to get informed about the exit-code + System.setSecurityManager(new SecurityManager() + { + public void checkExit(int status) + { + throw new SystemExitException(status); + } + + public void checkPermission(Permission perm) + { + } + + public void checkPermission(Permission perm, Object context) + { + } + }); + try + { + LoaderOptions.builder().parseArgs(args).build(); + } + catch (SystemExitException e) + { + assertEquals(expectedErrorCode, e.status); + } + finally + { + System.setSecurityManager(null); + } } }