natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into trunk

This commit is contained in:
Yifan Cai 2022-08-19 14:19:58 -07:00
parent 4526b3fcbd 83c169ec9e
commit 4aa3bbda79
8 changed files with 356 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

18
doc/modules/cassandra/pages/operating/bulk_loading.adoc
View File

@ -80,13 +80,23 @@ The following options are supported, with `-d,--nodes <initial hosts>` required:
-d,--nodes <initial hosts> Required.
Try to connect to these hosts (comma separated) initially for ring information
--entire-sstable-throttle-mib <throttle-mib> Entire SSTable throttle
speed in MiB/s (default 0 for unlimited).
--entire-sstable-inter-dc-throttle-mib <inter-dc-throttle-mib>
Entire SSTable inter-datacenter throttle
speed in MiB/s (default 0 for unlimited).
-f,--conf-path <path to config file> cassandra.yaml file path for streaming throughput and client/server SSL.
-h,--help Display this help message
-i,--ignore <NODES> Don't stream to this (comma separated) list of nodes
-idct,--inter-dc-throttle <inter-dc-throttle> Inter-datacenter throttle speed in Mbits (default unlimited)
-idct,--inter-dc-throttle <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-dc-throttle-mib> Inter-datacenter throttle speed in MiB/s (default 0 for unlimited)
-k,--target-keyspace <target keyspace name> Target
keyspace name
@ -111,8 +121,10 @@ The following options are supported, with `-d,--nodes <initial hosts>` required:
for TLS internode communication (default 7001)
-st,--store-type <STORE-TYPE> Client SSL:
type of store
-t,--throttle <throttle> Throttle
speed in Mbits (default unlimited)
-t,--throttle <throttle> (deprecated) Throttle speed in Mbits (default 0 for unlimited).
Use --throttle-mib instead.
--throttle-mib <throttle-mib> Throttle
speed in MiB/s (default 0 for unlimited)
-ts,--truststore <TRUSTSTORE> Client SSL:
full path to truststore
-tspw,--truststore-password <TRUSTSTORE-PASSWORD> Client SSL:

19
doc/modules/cassandra/pages/tools/sstable/sstableloader.adoc
View File

@ -45,10 +45,23 @@ internode communication (default 7001)
|--no-progress |don't display progress
|-t, --throttle <throttle> |throttle speed in Mbits (default unlimited)
|-t, --throttle <throttle> |(deprecated) throttle speed in Mbits
(default 0 for unlimited) Use --throttle-mib instead
|-idct, --inter-dc-throttle <inter-dc-throttle> |inter-datacenter
throttle speed in Mbits (default unlimited)
|--throttle-mib <throttle-mib> |throttle speed in MiB/s
(default 0 for unlimited)
|-idct, --inter-dc-throttle <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-dc-throttle-mib> |inter-datacenter
throttle speed in MiB/s (default 0 for unlimited)
|--entire-sstable-throttle-mib <throttle-mib> |entire SSTable throttle
speed in MiB/s (default 0 for unlimited)
|--entire-sstable-inter-dc-throttle-mib <inter-dc-throttle-mib> |entire
SSTable inter-datacenter throttle speed in MiB/s (default 0 for unlimited)
|-cph, --connections-per-host <connectionsPerHost> |number of concurrent
connections-per-host

10
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -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);

30
src/java/org/apache/cassandra/tools/BulkLoader.java
View File

@ -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);
}
}

161
src/java/org/apache/cassandra/tools/LoaderOptions.java
View File

@ -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");

47
test/unit/org/apache/cassandra/tools/BulkLoaderTest.java
View File

@ -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();
}
}

154
test/unit/org/apache/cassandra/tools/LoaderOptionsTest.java
View File

@ -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);
}
}
}