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:
Ekaterina Dimitrova 2022-08-01 20:39:44 -04:00
parent 0daf21244f dd08314ed6
commit 62ac5da78a
37 changed files with 1036 additions and 300 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
CHANGES.txt
View File

@ -21,6 +21,23 @@
* 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:
* Uncomment prepared_statements_cache_size, key_cache_size, counter_cache_size, index_summary_capacity which were
commented out by mistake in a previous patch
Fix breaking change with cache_load_timeout; cache_load_timeout_seconds <=0 and cache_load_timeout=0 are equivalent
and they both mean disabled
Deprecate public method setRate(final double throughputMbPerSec) in Compaction Manager in favor of
setRateInBytes(final double throughputBytesPerSec)
Revert breaking change removal of StressCQLSSTableWriter.Builder.withBufferSizeInMB(int size). Deprecate it in favor
of StressCQLSSTableWriter.Builder.withBufferSizeInMiB(int size)
Fix precision issues, add new -m flag (for nodetool/setstreamthroughput, nodetool/setinterdcstreamthroughput,
nodetool/getstreamthroughput and nodetoo/getinterdcstreamthroughput), add new -d flags (nodetool/getstreamthroughput, nodetool/getinterdcstreamthroughput, nodetool/getcompactionthroughput)
Fix a bug with precision in nodetool/compactionstats
Deprecate StorageService methods and add new ones for stream_throughput_outbound, inter_dc_stream_throughput_outbound,
compaction_throughput_outbound in the JMX MBean `org.apache.cassandra.db:type=StorageService`
Removed getEntireSSTableStreamThroughputMebibytesPerSec in favor of new getEntireSSTableStreamThroughputMebibytesPerSecAsDouble
in the JMX MBean `org.apache.cassandra.db:type=StorageService`
Removed getEntireSSTableInterDCStreamThroughputMebibytesPerSec in favor of getEntireSSTableInterDCStreamThroughputMebibytesPerSecAsDouble
in the JMX MBean `org.apache.cassandra.db:type=StorageService` (CASSANDRA-17725)
* Fix sstable_preemptive_open_interval disabled value. sstable_preemptive_open_interval = null backward compatible with
sstable_preemptive_open_interval_in_mb = -1 (CASSANDRA-17737)
* Remove usages of Path#toFile() in the snapshot apparatus (CASSANDRA-17769)

17
NEWS.txt
View File

@ -198,6 +198,7 @@ New features
Upgrading
---------
- `cache_load_timeout_seconds` being negative for disabled is equivalent to `cache_load_timeout` = 0 for disabled.
- `sstable_preemptive_open_interval_in_mb` being negative for disabled is equivalent to `sstable_preemptive_open_interval`
being null again. In the JMX MBean `org.apache.cassandra.db:type=StorageService`, the setter method
`setSSTablePreemptiveOpenIntervalInMB`still takes `intervalInMB` negative numbers for disabled.
@ -266,7 +267,21 @@ Upgrading
Deprecation
-----------
- `withBufferSizeInMB(int size)` in CQLSSTableWriter.Builder class is deprecated in favor of withBufferSizeInMiB(int size)
- In the JMX MBean `org.apache.cassandra.db:type=StorageService`: deprecate getter method `getStreamThroughputMbitPerSec`
in favor of getter method `getStreamThroughputMbitPerSecAsDouble`; deprecate getter method `getStreamThroughputMbPerSec`
in favor of getter methods `getStreamThroughputMebibytesPerSec` and `getStreamThroughputMebibytesPerSecAsDouble`;
deprecate getter method `getInterDCStreamThroughputMbitPerSec` in favor of getter method `getInterDCStreamThroughputMbitPerSecAsDouble`;
deprecate getter method `getInterDCStreamThroughputMbPerSec` in favor of getter methods `getInterDCStreamThroughputMebibytesPerSecAsDouble`;
deprecate getter method `getCompactionThroughputMbPerSec` in favor of getter methods `getCompactionThroughtputMibPerSecAsDouble`
and `getCompactionThroughtputBytesPerSec`; deprecate setter methods `setStreamThroughputMbPerSec` and `setStreamThroughputMbitPerSec`
in favor of `setStreamThroughputMebibytesPerSec`; deprecate setter methods `setInterDCStreamThroughputMbitPerSec` and
`setInterDCStreamThroughputMbPerSec` in favor of `setInterDCStreamThroughputMebibytesPerSec`. See CASSANDRA-17725 for further details.
- Deprecate public method `setRate(final double throughputMbPerSec)` in `Compaction Manager` in favor of
`setRateInBytes(final double throughputBytesPerSec)`
- `withBufferSizeInMB(int size)` in `StressCQLSSTableWriter.Builder` class is deprecated in favor of `withBufferSizeInMiB(int size)`
No change of functionality in the new one, only name change for clarity in regards to units and to follow naming
standartization.
- `withBufferSizeInMB(int size)` in `CQLSSTableWriter.Builder` class is deprecated in favor of `withBufferSizeInMiB(int size)`
No change of functionality in the new one, only name change for clarity in regards to units and to follow naming
standartization.
- The properties `keyspace_count_warn_threshold` and `table_count_warn_threshold` in cassandra.yaml have been

16
conf/cassandra.yaml
View File

@ -64,7 +64,7 @@ hinted_handoff_enabled: true
# Min unit: ms
max_hint_window: 3h
# Maximum throttle in KBs per second, per delivery thread. This will be
# Maximum throttle in KiBs per second, per delivery thread. This will be
# reduced proportionally to the number of nodes in the cluster. (If there
# are two nodes in the cluster, each delivery thread will use the maximum
# rate; if there are three, each will throttle to half of the maximum,
@ -86,7 +86,7 @@ max_hints_delivery_threads: 2
# Min unit: ms
hints_flush_period: 10000ms
# Maximum size for a single hints file, in megabytes.
# Maximum size for a single hints file, in mebibytes.
# Min unit: MiB
max_hints_file_size: 128MiB
@ -121,7 +121,7 @@ auto_hints_cleanup_enabled: false
#
# hint_window_persistent_enabled: true
# Maximum throttle in KBs per second, total. This will be
# Maximum throttle in KiBs per second, total. This will be
# reduced proportionally to the number of nodes in the cluster.
# Min unit: KiB
batchlog_replay_throttle: 1024KiB
@ -374,7 +374,7 @@ commit_failure_policy: stop
#
# Default value ("auto") is 1/256th of the heap or 10MiB, whichever is greater
# Min unit: MiB
# prepared_statements_cache_size:
prepared_statements_cache_size:
# Maximum size of the key cache in memory.
#
@ -389,7 +389,7 @@ commit_failure_policy: stop
#
# Default value is empty to make it "auto" (min(5% of Heap (in MiB), 100MiB)). Set to 0 to disable key cache.
# Min unit: MiB
# key_cache_size:
key_cache_size:
# Duration in seconds after which Cassandra should
# save the key cache. Caches are saved to saved_caches_directory as
@ -457,7 +457,7 @@ row_cache_save_period: 0s
# Default value is empty to make it "auto" (min(2.5% of Heap (in MiB), 50MiB)). Set to 0 to disable counter cache.
# NOTE: if you perform counter deletes and rely on low gcgs, you should disable the counter cache.
# Min unit: MiB
# counter_cache_size:
counter_cache_size:
# Duration in seconds after which Cassandra should
# save the counter cache (keys only). Caches are saved to saved_caches_directory as
@ -659,7 +659,7 @@ memtable_allocation_type: heap_buffers
# is 1/16th of the available heap. The main tradeoff is that smaller trees
# have less resolution, which can lead to over-streaming data. If you see heap
# pressure during repairs, consider lowering this, but you cannot go below
# one megabyte. If you see lots of over-streaming, consider raising
# one mebibyte. If you see lots of over-streaming, consider raising
# this or using subrange repair.
#
# For more details see https://issues.apache.org/jira/browse/CASSANDRA-14096.
@ -731,7 +731,7 @@ memtable_allocation_type: heap_buffers
# is a best-effort process. In extreme conditions Cassandra may need to use
# more than this amount of memory.
# Min unit: KiB
# index_summary_capacity:
index_summary_capacity:
# How frequently index summaries should be resampled. This is done
# periodically to redistribute memory from the fixed-size pool to sstables

16
src/java/org/apache/cassandra/config/Config.java
View File

@ -317,7 +317,7 @@ public class Config
public Integer unlogged_batch_across_partitions_warn_threshold = 10;
public volatile Integer concurrent_compactors;
@Replaces(oldName = "compaction_throughput_mb_per_sec", converter = Converters.MEBIBYTES_PER_SECOND_DATA_RATE, deprecated = true)
public volatile DataRateSpec.IntMebibytesPerSecondBound compaction_throughput = new DataRateSpec.IntMebibytesPerSecondBound("16MiB/s");
public volatile DataRateSpec.LongBytesPerSecondBound compaction_throughput = new DataRateSpec.LongBytesPerSecondBound("16MiB/s");
@Replaces(oldName = "compaction_large_partition_warning_threshold_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
public volatile DataStorageSpec.IntMebibytesBound compaction_large_partition_warning_threshold = new DataStorageSpec.IntMebibytesBound("100MiB");
@Replaces(oldName = "min_free_space_per_drive_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
@ -333,13 +333,13 @@ public class Config
@Deprecated
public int max_streaming_retries = 3;
@Replaces(oldName = "stream_throughput_outbound_megabits_per_sec", converter = Converters.MEGABITS_TO_MEBIBYTES_PER_SECOND_DATA_RATE, deprecated = true)
public volatile DataRateSpec.IntMebibytesPerSecondBound stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s");
@Replaces(oldName = "inter_dc_stream_throughput_outbound_megabits_per_sec", converter = Converters.MEGABITS_TO_MEBIBYTES_PER_SECOND_DATA_RATE, deprecated = true)
public volatile DataRateSpec.IntMebibytesPerSecondBound inter_dc_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s");
@Replaces(oldName = "stream_throughput_outbound_megabits_per_sec", converter = Converters.MEGABITS_TO_BYTES_PER_SECOND_DATA_RATE, deprecated = true)
public volatile DataRateSpec.LongBytesPerSecondBound stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound("24MiB/s");
@Replaces(oldName = "inter_dc_stream_throughput_outbound_megabits_per_sec", converter = Converters.MEGABITS_TO_BYTES_PER_SECOND_DATA_RATE, deprecated = true)
public volatile DataRateSpec.LongBytesPerSecondBound inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound("24MiB/s");
public volatile DataRateSpec.IntMebibytesPerSecondBound entire_sstable_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s");
public volatile DataRateSpec.IntMebibytesPerSecondBound entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s");
public volatile DataRateSpec.LongBytesPerSecondBound entire_sstable_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound("24MiB/s");
public volatile DataRateSpec.LongBytesPerSecondBound entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound("24MiB/s");
public String[] data_file_directories = new String[0];
@ -454,7 +454,7 @@ public class Config
public DataStorageSpec.LongMebibytesBound paxos_cache_size = null;
@Replaces(oldName = "cache_load_timeout_seconds", converter = Converters.SECONDS_DURATION, deprecated = true)
@Replaces(oldName = "cache_load_timeout_seconds", converter = Converters.NEGATIVE_SECONDS_DURATION, deprecated = true)
public DurationSpec.IntSecondsBound cache_load_timeout = new DurationSpec.IntSecondsBound("30s");
private static boolean isClientMode = false;

14
src/java/org/apache/cassandra/config/Converters.java
View File

@ -21,6 +21,8 @@ package org.apache.cassandra.config;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND;
/**
* Converters for backward compatibility with the old cassandra.yaml where duration, data rate and
* data storage configuration parameters were provided only by value and the expected unit was part of the configuration
@ -100,16 +102,16 @@ public enum Converters
BYTES_CUSTOM_DATASTORAGE(Long.class, DataStorageSpec.LongBytesBound.class,
o -> o == -1 ? null : new DataStorageSpec.LongBytesBound(o),
o -> o == null ? null : o.toBytes()),
MEBIBYTES_PER_SECOND_DATA_RATE(Integer.class, DataRateSpec.IntMebibytesPerSecondBound.class,
DataRateSpec.IntMebibytesPerSecondBound::new,
MEBIBYTES_PER_SECOND_DATA_RATE(Integer.class, DataRateSpec.LongBytesPerSecondBound.class,
i -> new DataRateSpec.LongBytesPerSecondBound(i, MEBIBYTES_PER_SECOND),
o -> o == null ? null : o.toMebibytesPerSecondAsInt()),
/**
* This converter is a custom one to support backward compatibility for stream_throughput_outbound and
* inter_dc_stream_throughput_outbound which were provided in megatibs per second prior CASSANDRA-15234.
* inter_dc_stream_throughput_outbound which were provided in megabits per second prior CASSANDRA-15234.
*/
MEGABITS_TO_MEBIBYTES_PER_SECOND_DATA_RATE(Integer.class, DataRateSpec.IntMebibytesPerSecondBound.class,
i -> DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(i),
o -> o == null ? null : o.toMegabitsPerSecondAsInt());
MEGABITS_TO_BYTES_PER_SECOND_DATA_RATE(Integer.class, DataRateSpec.LongBytesPerSecondBound.class,
i -> DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(i),
o -> o == null ? null : o.toMegabitsPerSecondAsInt());
private final Class<?> oldType;
private final Class<?> newType;
private final Function<Object, Object> convert;

66
src/java/org/apache/cassandra/config/DataRateSpec.java
View File

@ -23,10 +23,10 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import com.google.common.math.DoubleMath;
import com.google.common.primitives.Ints;
import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.BYTES_PER_SECOND;
import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND;
/**
* Represents a data rate type used for cassandra configuration. It supports the opportunity for the users to be able to
@ -39,7 +39,7 @@ public abstract class DataRateSpec
*/
private static final Pattern UNITS_PATTERN = Pattern.compile("^(\\d+)(MiB/s|KiB/s|B/s)$");
private final double quantity;
private final long quantity;
private final DataRateUnit unit;
@ -52,7 +52,7 @@ public abstract class DataRateSpec
throw new IllegalArgumentException("Invalid data rate: " + value + " Accepted units: MiB/s, KiB/s, B/s where " +
"case matters and " + "only non-negative values are valid");
quantity = (double) Long.parseLong(matcher.group(1));
quantity = Long.parseLong(matcher.group(1));
unit = DataRateUnit.fromSymbol(matcher.group(2));
}
@ -63,7 +63,7 @@ public abstract class DataRateSpec
validateQuantity(value, quantity(), unit(), minUnit, max);
}
private DataRateSpec(double quantity, DataRateUnit unit, DataRateUnit minUnit, long max)
private DataRateSpec(long quantity, DataRateUnit unit, DataRateUnit minUnit, long max)
{
this.quantity = quantity;
this.unit = unit;
@ -212,7 +212,7 @@ public abstract class DataRateSpec
@Override
public String toString()
{
return Math.round(quantity) + unit.symbol;
return (DoubleMath.isMathematicalInteger(quantity) ? (long) quantity : quantity) + unit.symbol;
}
/**
@ -238,7 +238,7 @@ public abstract class DataRateSpec
* @param quantity where quantity shouldn't be bigger than Long.MAX_VALUE - 1 in bytes per second
* @param unit in which the provided quantity is
*/
public LongBytesPerSecondBound(double quantity, DataRateUnit unit)
public LongBytesPerSecondBound(long quantity, DataRateUnit unit)
{
super(quantity, unit, BYTES_PER_SECOND, Long.MAX_VALUE);
}
@ -252,59 +252,21 @@ public abstract class DataRateSpec
{
this(bytesPerSecond, BYTES_PER_SECOND);
}
}
/**
* Represents a data rate used for Cassandra configuration. The bound is [0, Integer.MAX_VALUE) in mebibytes per second.
* If the user sets a different unit - we still validate that converted to mebibytes per second the quantity will not exceed
* that upper bound. (CASSANDRA-17571)
*/
public final static class IntMebibytesPerSecondBound extends DataRateSpec
{
/**
* Creates a {@code DataRateSpec.IntMebibytesPerSecondBound} of the specified amount with bound [0, Integer.MAX_VALUE) mebibytes per second.
*
* @param value the data rate
*/
public IntMebibytesPerSecondBound(String value)
{
super(value, MEBIBYTES_PER_SECOND, Integer.MAX_VALUE);
}
/**
* Creates a {@code DataRateSpec.IntMebibytesPerSecondBound} of the specified amount in the specified unit.
*
* @param quantity where quantity shouldn't be bigger than Integer.MAX_VALUE - 1 in mebibytes per second
* @param unit in which the provided quantity is
*/
public IntMebibytesPerSecondBound(double quantity, DataRateUnit unit)
{
super(quantity, unit, MEBIBYTES_PER_SECOND, Integer.MAX_VALUE);
}
/**
* Creates a {@code DataRateSpec.IntMebibytesPerSecondBound} of the specified amount in mebibytes per second.
*
* @param mebibytesPerSecond where mebibytesPerSecond shouldn't be bigger than Long.MAX_VALUE-1
*/
public IntMebibytesPerSecondBound(long mebibytesPerSecond)
{
this (mebibytesPerSecond, MEBIBYTES_PER_SECOND);
}
// this one should be used only for backward compatibility for stream_throughput_outbound and inter_dc_stream_throughput_outbound
// which were in megabits per second in 4.0. Do not start using it for any new properties
public static IntMebibytesPerSecondBound megabitsPerSecondInMebibytesPerSecond(long megabitsPerSecond)
@Deprecated
public static LongBytesPerSecondBound megabitsPerSecondInBytesPerSecond(long megabitsPerSecond)
{
final double MEBIBYTES_PER_MEGABIT = 0.119209289550781;
double mebibytesPerSecond = (double) megabitsPerSecond * MEBIBYTES_PER_MEGABIT;
final long BYTES_PER_MEGABIT = 125_000;
long bytesPerSecond = megabitsPerSecond * BYTES_PER_MEGABIT;
if (megabitsPerSecond >= Integer.MAX_VALUE)
throw new IllegalArgumentException("Invalid data rate: " + megabitsPerSecond + " megabits per second; " +
"stream_throughput_outbound and inter_dc_stream_throughput_outbound" +
" should be between 0 and " + Integer.MAX_VALUE + " in megabits per second");
"stream_throughput_outbound and inter_dc_stream_throughput_outbound" +
" should be between 0 and " + (Integer.MAX_VALUE - 1) + " in megabits per second");
return new IntMebibytesPerSecondBound(mebibytesPerSecond, MEBIBYTES_PER_SECOND);
return new LongBytesPerSecondBound(bytesPerSecond, BYTES_PER_SECOND);
}
}
@ -385,7 +347,7 @@ public abstract class DataRateSpec
{
if (d > MAX / (MEGABITS_PER_MEBIBYTE))
return MAX;
return Math.round(d * MEGABITS_PER_MEBIBYTE);
return d * MEGABITS_PER_MEBIBYTE;
}
public double convert(double source, DataRateUnit sourceUnit)

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

@ -93,6 +93,8 @@ import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.config.CassandraRelevantProperties.OS_ARCH;
import static org.apache.cassandra.config.CassandraRelevantProperties.SUN_ARCH_DATA_MODEL;
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_JVM_DTEST_DISABLE_SSL;
import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.BYTES_PER_SECOND;
import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND;
import static org.apache.cassandra.config.DataStorageSpec.DataStorageUnit.MEBIBYTES;
import static org.apache.cassandra.io.util.FileUtils.ONE_GIB;
import static org.apache.cassandra.io.util.FileUtils.ONE_MIB;
@ -398,16 +400,7 @@ public class DatabaseDescriptor
//InetAddressAndPort and get the right defaults
InetAddressAndPort.initializeDefaultPort(getStoragePort());
// below 2 checks are needed in order to match the pre-CASSANDRA-15234 upper bound for those parameters which were still in megabits per second
if (conf.stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of stream_throughput_outbound: " + conf.stream_throughput_outbound.toString(), false);
}
if (conf.inter_dc_stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of inter_dc_stream_throughput_outbound: " + conf.inter_dc_stream_throughput_outbound.toString(), false);
}
validateUpperBoundStreamingConfig();
if (conf.auto_snapshot_ttl != null)
{
@ -915,6 +908,36 @@ public class DatabaseDescriptor
logInitializationOutcome(logger);
}
@VisibleForTesting
static void validateUpperBoundStreamingConfig() throws ConfigurationException
{
// below 2 checks are needed in order to match the pre-CASSANDRA-15234 upper bound for those parameters which were still in megabits per second
if (conf.stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of stream_throughput_outbound: " + conf.stream_throughput_outbound.toString(), false);
}
if (conf.inter_dc_stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of inter_dc_stream_throughput_outbound: " + conf.inter_dc_stream_throughput_outbound.toString(), false);
}
if (conf.entire_sstable_stream_throughput_outbound.toMebibytesPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of entire_sstable_stream_throughput_outbound: " + conf.entire_sstable_stream_throughput_outbound.toString(), false);
}
if (conf.entire_sstable_inter_dc_stream_throughput_outbound.toMebibytesPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of entire_sstable_inter_dc_stream_throughput_outbound: " + conf.entire_sstable_inter_dc_stream_throughput_outbound.toString(), false);
}
if (conf.compaction_throughput.toMebibytesPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of compaction_throughput: " + conf.compaction_throughput.toString(), false);
}
}
@VisibleForTesting
static void applyConcurrentValidations(Config config)
{
@ -1962,14 +1985,35 @@ public class DatabaseDescriptor
return conf.compaction_throughput.toMebibytesPerSecondAsInt();
}
public static double getCompactionThroughputBytesPerSec()
{
return conf.compaction_throughput.toBytesPerSecond();
}
public static double getCompactionThroughputMebibytesPerSec()
{
return conf.compaction_throughput.toMebibytesPerSecond();
}
@VisibleForTesting // only for testing!
public static void setCompactionThroughputBytesPerSec(int value)
{
if (BYTES_PER_SECOND.toMebibytesPerSecond(value) >= Integer.MAX_VALUE)
throw new IllegalArgumentException("compaction_throughput: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in MiB/s");
conf.compaction_throughput = new DataRateSpec.LongBytesPerSecondBound(value);
}
public static void setCompactionThroughputMebibytesPerSec(int value)
{
conf.compaction_throughput = new DataRateSpec.IntMebibytesPerSecondBound(value);
if (value == Integer.MAX_VALUE)
throw new IllegalArgumentException("compaction_throughput: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in MiB/s");
conf.compaction_throughput = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND);
}
public static long getCompactionLargePartitionWarningThreshold() { return conf.compaction_large_partition_warning_threshold.toBytesInLong(); }
@ -2025,19 +2069,39 @@ public class DatabaseDescriptor
return conf.stream_throughput_outbound.toMegabitsPerSecondAsInt();
}
public static double getStreamThroughputOutboundMegabitsPerSecAsDouble()
{
return conf.stream_throughput_outbound.toMegabitsPerSecond();
}
public static double getStreamThroughputOutboundMebibytesPerSec()
{
return conf.stream_throughput_outbound.toMebibytesPerSecond();
}
public static void setStreamThroughputOutboundMegabitsPerSec(int value)
public static double getStreamThroughputOutboundBytesPerSec()
{
conf.stream_throughput_outbound = DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(value);
return conf.stream_throughput_outbound.toBytesPerSecond();
}
public static int getEntireSSTableStreamThroughputOutboundMebibytesPerSecAsInt()
public static int getStreamThroughputOutboundMebibytesPerSecAsInt()
{
return conf.entire_sstable_stream_throughput_outbound.toMebibytesPerSecondAsInt();
return conf.stream_throughput_outbound.toMebibytesPerSecondAsInt();
}
public static void setStreamThroughputOutboundMebibytesPerSecAsInt(int value)
{
if (MEBIBYTES_PER_SECOND.toMegabitsPerSecond(value) >= Integer.MAX_VALUE)
throw new IllegalArgumentException("stream_throughput_outbound: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in megabits/s");
conf.stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND);
}
public static void setStreamThroughputOutboundMegabitsPerSec(int value)
{
conf.stream_throughput_outbound = DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(value);
}
public static double getEntireSSTableStreamThroughputOutboundMebibytesPerSec()
@ -2045,9 +2109,19 @@ public class DatabaseDescriptor
return conf.entire_sstable_stream_throughput_outbound.toMebibytesPerSecond();
}
public static double getEntireSSTableStreamThroughputOutboundBytesPerSec()
{
return conf.entire_sstable_stream_throughput_outbound.toBytesPerSecond();
}
public static void setEntireSSTableStreamThroughputOutboundMebibytesPerSec(int value)
{
conf.entire_sstable_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(value);
if (value == Integer.MAX_VALUE)
throw new IllegalArgumentException("entire_sstable_stream_throughput_outbound: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in MiB/s");
conf.entire_sstable_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND);
}
public static int getInterDCStreamThroughputOutboundMegabitsPerSec()
@ -2055,14 +2129,44 @@ public class DatabaseDescriptor
return conf.inter_dc_stream_throughput_outbound.toMegabitsPerSecondAsInt();
}
public static double getInterDCStreamThroughputOutboundMegabitsPerSecAsDouble()
{
return conf.inter_dc_stream_throughput_outbound.toMegabitsPerSecond();
}
public static double getInterDCStreamThroughputOutboundMebibytesPerSec()
{
return conf.inter_dc_stream_throughput_outbound.toMebibytesPerSecond();
}
public static double getInterDCStreamThroughputOutboundBytesPerSec()
{
return conf.inter_dc_stream_throughput_outbound.toBytesPerSecond();
}
public static int getInterDCStreamThroughputOutboundMebibytesPerSecAsInt()
{
return conf.inter_dc_stream_throughput_outbound.toMebibytesPerSecondAsInt();
}
public static void setInterDCStreamThroughputOutboundMebibytesPerSecAsInt(int value)
{
if (MEBIBYTES_PER_SECOND.toMegabitsPerSecond(value) >= Integer.MAX_VALUE)
throw new IllegalArgumentException("inter_dc_stream_throughput_outbound: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in megabits/s");
conf.inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND);
}
public static void setInterDCStreamThroughputOutboundMegabitsPerSec(int value)
{
conf.inter_dc_stream_throughput_outbound = DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(value);
conf.inter_dc_stream_throughput_outbound = DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(value);
}
public static double getEntireSSTableInterDCStreamThroughputOutboundBytesPerSec()
{
return conf.entire_sstable_inter_dc_stream_throughput_outbound.toBytesPerSecond();
}
public static double getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec()
@ -2070,14 +2174,14 @@ public class DatabaseDescriptor
return conf.entire_sstable_inter_dc_stream_throughput_outbound.toMebibytesPerSecond();
}
public static int getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSecAsInt()
{
return conf.entire_sstable_inter_dc_stream_throughput_outbound.toMebibytesPerSecondAsInt();
}
public static void setEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec(int value)
{
conf.entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(value);
if (value == Integer.MAX_VALUE)
throw new IllegalArgumentException("entire_sstable_inter_dc_stream_throughput_outbound: " + value +
" is too large; it should be less than " +
Integer.MAX_VALUE + " in MiB/s");
conf.entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(value, MEBIBYTES_PER_SECOND);
}
/**

20
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -153,18 +153,30 @@ public class CompactionManager implements CompactionManagerMBean
*/
public RateLimiter getRateLimiter()
{
setRate(DatabaseDescriptor.getCompactionThroughputMebibytesPerSec());
setRateInBytes(DatabaseDescriptor.getCompactionThroughputBytesPerSec());
return compactionRateLimiter;
}
/**
* Sets the rate for the rate limiter. When compaction_throughput is 0 or node is bootstrapping,
* this sets the rate to Double.MAX_VALUE bytes per second.
* @param throughPutMiBPerSec throughput to set in MiB/s
* @param throughputMbPerSec throughput to set in MiB/s
* @deprecated Use setRateInBytes instead
*/
public void setRate(final double throughPutMiBPerSec)
@Deprecated
public void setRate(final double throughputMbPerSec)
{
double throughput = throughPutMiBPerSec * 1024.0 * 1024.0;
setRateInBytes(throughputMbPerSec * 1024.0 * 1024);
}
/**
* Sets the rate for the rate limiter. When compaction_throughput is 0 or node is bootstrapping,
* this sets the rate to Double.MAX_VALUE bytes per second.
* @param throughputBytesPerSec throughput to set in B/s
*/
public void setRateInBytes(final double throughputBytesPerSec)
{
double throughput = throughputBytesPerSec;
// if throughput is set to 0, throttling is disabled
if (throughput == 0 || StorageService.instance.isBootstrapMode())
throughput = Double.MAX_VALUE;

82
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1525,36 +1525,60 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public void setStreamThroughputMbitPerSec(int value)
{
int oldValue = DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec();
double oldValue = DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSecAsDouble();
DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(value);
StreamManager.StreamRateLimiter.updateThroughput();
logger.info("setstreamthroughput: throttle set to {}{} megabits per second (was {} megabits per second)",
logger.info("setstreamthroughput: throttle set to {}{} megabits per second (was approximately {} megabits per second)",
value, value <= 0 ? " (unlimited)" : "", oldValue);
}
public void setStreamThroughputMebibytesPerSec(int value)
{
double oldValue = DatabaseDescriptor.getStreamThroughputOutboundMebibytesPerSec();
DatabaseDescriptor.setStreamThroughputOutboundMebibytesPerSecAsInt(value);
StreamManager.StreamRateLimiter.updateThroughput();
logger.info("setstreamthroughput: throttle set to {}{} MiB/s (was {} MiB/s)", value, value <= 0 ? " (unlimited)" : "", oldValue);
}
public double getStreamThroughputMebibytesPerSecAsDouble()
{
return DatabaseDescriptor.getStreamThroughputOutboundMebibytesPerSec();
}
public int getStreamThroughputMebibytesPerSec()
{
return DatabaseDescriptor.getStreamThroughputOutboundMebibytesPerSecAsInt();
}
@Deprecated
public int getStreamThroughputMbPerSec()
{
return getStreamThroughputMbitPerSec();
}
@Deprecated
public int getStreamThroughputMbitPerSec()
{
return DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec();
}
public double getStreamThroughputMbitPerSecAsDouble()
{
return DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSecAsDouble();
}
public void setEntireSSTableStreamThroughputMebibytesPerSec(int value)
{
int oldValue = DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSecAsInt();
double oldValue = DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec();
DatabaseDescriptor.setEntireSSTableStreamThroughputOutboundMebibytesPerSec(value);
StreamManager.StreamRateLimiter.updateEntireSSTableThroughput();
logger.info("setstreamthroughput (entire SSTable): throttle set to {}{} MiB/s (was {} MiB/s)",
value, value <= 0 ? " (unlimited)" : "", oldValue);
}
public int getEntireSSTableStreamThroughputMebibytesPerSec()
public double getEntireSSTableStreamThroughputMebibytesPerSecAsDouble()
{
return DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSecAsInt();
return DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec();
}
@Deprecated
@ -1565,7 +1589,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public void setInterDCStreamThroughputMbitPerSec(int value)
{
int oldValue = DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec();
double oldValue = DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSecAsDouble();
DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(value);
StreamManager.StreamRateLimiter.updateInterDCThroughput();
logger.info("setinterdcstreamthroughput: throttle set to {}{} megabits per second (was {} megabits per second)", value, value <= 0 ? " (unlimited)" : "", oldValue);
@ -1577,24 +1601,59 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return getInterDCStreamThroughputMbitPerSec();
}
@Deprecated
public int getInterDCStreamThroughputMbitPerSec()
{
return DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec();
}
public double getInterDCStreamThroughputMbitPerSecAsDouble()
{
return DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSecAsDouble();
}
public void setInterDCStreamThroughputMebibytesPerSec(int value)
{
double oldValue = DatabaseDescriptor.getInterDCStreamThroughputOutboundMebibytesPerSec();
DatabaseDescriptor.setInterDCStreamThroughputOutboundMebibytesPerSecAsInt(value);
StreamManager.StreamRateLimiter.updateInterDCThroughput();
logger.info("setinterdcstreamthroughput: throttle set to {}{} MiB/s (was {} MiB/s)", value, value <= 0 ? " (unlimited)" : "", oldValue);
}
public int getInterDCStreamThroughputMebibytesPerSec()
{
return DatabaseDescriptor.getInterDCStreamThroughputOutboundMebibytesPerSecAsInt();
}
public double getInterDCStreamThroughputMebibytesPerSecAsDouble()
{
return DatabaseDescriptor.getInterDCStreamThroughputOutboundMebibytesPerSec();
}
public void setEntireSSTableInterDCStreamThroughputMebibytesPerSec(int value)
{
int oldValue = DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSecAsInt();
double oldValue = DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec();
DatabaseDescriptor.setEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec(value);
StreamManager.StreamRateLimiter.updateEntireSSTableInterDCThroughput();
logger.info("setinterdcstreamthroughput (entire SSTable): throttle set to {}{} MiB/s (was {} MiB/s)", value, value <= 0 ? " (unlimited)" : "", oldValue);
}
public int getEntireSSTableInterDCStreamThroughputMebibytesPerSec()
public double getEntireSSTableInterDCStreamThroughputMebibytesPerSecAsDouble()
{
return DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSecAsInt();
return DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec();
}
public double getCompactionThroughtputMibPerSecAsDouble()
{
return DatabaseDescriptor.getCompactionThroughputMebibytesPerSec();
}
public long getCompactionThroughtputBytesPerSec()
{
return (long)DatabaseDescriptor.getCompactionThroughputBytesPerSec();
}
@Deprecated
public int getCompactionThroughputMbPerSec()
{
return DatabaseDescriptor.getCompactionThroughputMebibytesPerSecAsInt();
@ -1602,9 +1661,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
public void setCompactionThroughputMbPerSec(int value)
{
int oldValue = DatabaseDescriptor.getCompactionThroughputMebibytesPerSecAsInt();
double oldValue = DatabaseDescriptor.getCompactionThroughputMebibytesPerSec();
DatabaseDescriptor.setCompactionThroughputMebibytesPerSec(value);
CompactionManager.instance.setRate(value);
double valueInBytes = value * 1024.0 * 1024.0;
CompactionManager.instance.setRateInBytes(valueInBytes);
logger.info("compactionthroughput: throttle set to {} mebibytes per second (was {} mebibytes per second)",
value, oldValue);
}

49
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -625,26 +625,71 @@ public interface StorageServiceMBean extends NotificationEmitter
public long getTruncateRpcTimeout();
public void setStreamThroughputMbitPerSec(int value);
/**
* @return stream_throughput_outbound in megabits
* @deprecated Use getStreamThroughputMbitPerSecAsDouble instead as this one will provide a rounded value
*/
@Deprecated
public int getStreamThroughputMbitPerSec();
public double getStreamThroughputMbitPerSecAsDouble();
@Deprecated
public void setStreamThroughputMbPerSec(int value);
/**
* @return stream_throughput_outbound in MiB
* @deprecated Use getStreamThroughputMebibytesPerSecAsDouble instead as this one will provide a rounded value
*/
@Deprecated
public int getStreamThroughputMbPerSec();
public void setStreamThroughputMebibytesPerSec(int value);
/**
* Below method returns stream_throughput_outbound rounded, for precise number, please, use getStreamThroughputMebibytesPerSecAsDouble
* @return stream_throughput_outbound in MiB
*/
public int getStreamThroughputMebibytesPerSec();
public double getStreamThroughputMebibytesPerSecAsDouble();
public void setInterDCStreamThroughputMbitPerSec(int value);
/**
* @return inter_dc_stream_throughput_outbound in megabits
* @deprecated Use getInterDCStreamThroughputMbitPerSecAsDouble instead as this one will provide a rounded value
*/
@Deprecated
public int getInterDCStreamThroughputMbitPerSec();
public double getInterDCStreamThroughputMbitPerSecAsDouble();
@Deprecated
public void setInterDCStreamThroughputMbPerSec(int value);
/**
* @return inter_dc_stream_throughput_outbound in MiB
* @deprecated Use getInterDCStreamThroughputMebibytesPerSecAsDouble instead as this one will provide a rounded value
*/
@Deprecated
public int getInterDCStreamThroughputMbPerSec();
public void setInterDCStreamThroughputMebibytesPerSec(int value);
/**
* Below method returns Inter_dc_stream_throughput_outbound rounded, for precise number, please, use getInterDCStreamThroughputMebibytesPerSecAsDouble
* @return inter_dc_stream_throughput_outbound in MiB
*/
public int getInterDCStreamThroughputMebibytesPerSec();
public double getInterDCStreamThroughputMebibytesPerSecAsDouble();
public void setEntireSSTableStreamThroughputMebibytesPerSec(int value);
public int getEntireSSTableStreamThroughputMebibytesPerSec();
public double getEntireSSTableStreamThroughputMebibytesPerSecAsDouble();
public void setEntireSSTableInterDCStreamThroughputMebibytesPerSec(int value);
public int getEntireSSTableInterDCStreamThroughputMebibytesPerSec();
public double getEntireSSTableInterDCStreamThroughputMebibytesPerSecAsDouble();
public double getCompactionThroughtputMibPerSecAsDouble();
public long getCompactionThroughtputBytesPerSec();
/**
* @return compaction_throughgput in MiB
* @deprecated Use getCompactionThroughtputMibPerSecAsDouble instead as this one will provide a rounded value
*/
@Deprecated
public int getCompactionThroughputMbPerSec();
public void setCompactionThroughputMbPerSec(int value);

19
src/java/org/apache/cassandra/streaming/StreamManager.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.streaming;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.management.ListenerNotFoundException;
import javax.management.MBeanNotificationInfo;
@ -71,8 +70,8 @@ public class StreamManager implements StreamManagerMBean
return new StreamRateLimiter(peer,
StreamRateLimiter.LIMITER,
StreamRateLimiter.INTER_DC_LIMITER,
DatabaseDescriptor.getStreamThroughputOutboundMebibytesPerSec(),
DatabaseDescriptor.getInterDCStreamThroughputOutboundMebibytesPerSec());
DatabaseDescriptor.getStreamThroughputOutboundBytesPerSec(),
DatabaseDescriptor.getInterDCStreamThroughputOutboundBytesPerSec());
}
/**
@ -90,8 +89,8 @@ public class StreamManager implements StreamManagerMBean
return new StreamRateLimiter(peer,
StreamRateLimiter.ENTIRE_SSTABLE_LIMITER,
StreamRateLimiter.ENTIRE_SSTABLE_INTER_DC_LIMITER,
DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec(),
DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec());
DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundBytesPerSec(),
DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundBytesPerSec());
}
public static class StreamRateLimiter implements StreamingDataOutputPlus.RateLimiter
@ -159,25 +158,25 @@ public class StreamManager implements StreamManagerMBean
private static double calculateRateInBytes()
{
double throughput = DatabaseDescriptor.getStreamThroughputOutboundMebibytesPerSec();
double throughput = DatabaseDescriptor.getStreamThroughputOutboundBytesPerSec();
return calculateEffectiveRateInBytes(throughput);
}
private static double calculateInterDCRateInBytes()
{
double throughput = DatabaseDescriptor.getInterDCStreamThroughputOutboundMebibytesPerSec();
double throughput = DatabaseDescriptor.getInterDCStreamThroughputOutboundBytesPerSec();
return calculateEffectiveRateInBytes(throughput);
}
private static double calculateEntireSSTableRateInBytes()
{
double throughput = DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundMebibytesPerSec();
double throughput = DatabaseDescriptor.getEntireSSTableStreamThroughputOutboundBytesPerSec();
return calculateEffectiveRateInBytes(throughput);
}
private static double calculateEntireSSTableInterDCRateInBytes()
{
double throughput = DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundMebibytesPerSec();
double throughput = DatabaseDescriptor.getEntireSSTableInterDCStreamThroughputOutboundBytesPerSec();
return calculateEffectiveRateInBytes(throughput);
}
@ -209,7 +208,7 @@ public class StreamManager implements StreamManagerMBean
{
// if throughput is set to 0, throttling is disabled
return throughput > 0
? throughput * BYTES_PER_MEBIBYTE
? throughput
: Double.MAX_VALUE;
}
}

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

@ -422,22 +422,32 @@ public class LoaderOptions
// below 2 checks are needed in order to match the pre-CASSANDRA-15234 upper bound for those parameters which were still in megabits per second
if (config.stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of stream_throughput_outbound: " + config.stream_throughput_outbound.toString(), false);
throw new ConfigurationException("stream_throughput_outbound: " + config.stream_throughput_outbound.toString() + " is too large", false);
}
if (config.inter_dc_stream_throughput_outbound.toMegabitsPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("Invalid value of inter_dc_stream_throughput_outbound: " + config.inter_dc_stream_throughput_outbound.toString(), false);
throw new ConfigurationException("inter_dc_stream_throughput_outbound: " + config.inter_dc_stream_throughput_outbound.toString() + " is too large", false);
}
if (config.entire_sstable_stream_throughput_outbound.toMebibytesPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("entire_sstable_stream_throughput_outbound: " + config.entire_sstable_stream_throughput_outbound.toString() + " is too large", false);
}
if (config.entire_sstable_inter_dc_stream_throughput_outbound.toMebibytesPerSecond() >= Integer.MAX_VALUE)
{
throw new ConfigurationException("entire_sstable_inter_dc_stream_throughput_outbound: " + config.entire_sstable_inter_dc_stream_throughput_outbound.toString() + " is too large", false);
}
}
else
{
config = new Config();
// unthrottle stream by default
config.stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(0);
config.inter_dc_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(0);
config.entire_sstable_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(0);
config.entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.IntMebibytesPerSecondBound(0);
config.stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(0);
config.inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(0);
config.entire_sstable_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(0);
config.entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(0);
}
if (cmd.hasOption(STORAGE_PORT_OPTION))

51
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -1253,11 +1253,22 @@ public class NodeProbe implements AutoCloseable
ssProxy.setCompactionThroughputMbPerSec(value);
}
@Deprecated
public int getCompactionThroughput()
{
return ssProxy.getCompactionThroughputMbPerSec();
}
public double getCompactionThroughputMebibytesAsDouble()
{
return ssProxy.getCompactionThroughtputMibPerSecAsDouble();
}
public long getCompactionThroughputBytes()
{
return ssProxy.getCompactionThroughtputBytesPerSec();
}
public void setBatchlogReplayThrottle(int value)
{
ssProxy.setBatchlogReplayThrottleInKB(value);
@ -1327,24 +1338,46 @@ public class NodeProbe implements AutoCloseable
}
}
@Deprecated
public int getStreamThroughput()
{
return ssProxy.getStreamThroughputMbitPerSec();
}
public double getStreamThroughputAsDouble()
{
return ssProxy.getStreamThroughputMbitPerSecAsDouble();
}
@Deprecated
public int getInterDCStreamThroughput()
{
return ssProxy.getInterDCStreamThroughputMbitPerSec();
}
public int getEntireSSTableStreamThroughput()
public double getInterDCStreamThroughputAsDouble()
{
return ssProxy.getEntireSSTableStreamThroughputMebibytesPerSec();
return ssProxy.getInterDCStreamThroughputMbitPerSecAsDouble();
}
public int getEntireSSTableInterDCStreamThroughput()
public double getStreamThroughputMibAsDouble()
{
return ssProxy.getEntireSSTableInterDCStreamThroughputMebibytesPerSec();
return ssProxy.getStreamThroughputMebibytesPerSecAsDouble();
}
public double getInterDCStreamThroughputMibAsDouble()
{
return ssProxy.getInterDCStreamThroughputMebibytesPerSecAsDouble();
}
public double getEntireSSTableStreamThroughput()
{
return ssProxy.getEntireSSTableStreamThroughputMebibytesPerSecAsDouble();
}
public double getEntireSSTableInterDCStreamThroughput()
{
return ssProxy.getEntireSSTableInterDCStreamThroughputMebibytesPerSecAsDouble();
}
public double getTraceProbability()
@ -1445,11 +1478,21 @@ public class NodeProbe implements AutoCloseable
ssProxy.setStreamThroughputMbitPerSec(value);
}
public void setStreamThroughputMiB(int value)
{
ssProxy.setStreamThroughputMebibytesPerSec(value);
}
public void setInterDCStreamThroughput(int value)
{
ssProxy.setInterDCStreamThroughputMbitPerSec(value);
}
public void setInterDCStreamThroughputMiB(int value)
{
ssProxy.setInterDCStreamThroughputMebibytesPerSec(value);
}
public void setEntireSSTableStreamThroughput(int value)
{
ssProxy.setEntireSSTableStreamThroughputMebibytesPerSec(value);

12
src/java/org/apache/cassandra/tools/nodetool/CompactionStats.java
View File

@ -75,15 +75,15 @@ public class CompactionStats extends NodeToolCmd
}
}
out.println();
reportCompactionTable(cm.getCompactions(), probe.getCompactionThroughput(), humanReadable, vtableOutput, out);
reportCompactionTable(cm.getCompactions(), probe.getCompactionThroughputBytes(), humanReadable, vtableOutput, out);
}
public static void reportCompactionTable(List<Map<String,String>> compactions, int compactionThroughput, boolean humanReadable, PrintStream out)
public static void reportCompactionTable(List<Map<String,String>> compactions, long compactionThroughputInBytes, boolean humanReadable, PrintStream out)
{
reportCompactionTable(compactions, compactionThroughput, humanReadable, false, out);
reportCompactionTable(compactions, compactionThroughputInBytes, humanReadable, false, out);
}
public static void reportCompactionTable(List<Map<String,String>> compactions, int compactionThroughput, boolean humanReadable, boolean vtableOutput, PrintStream out)
public static void reportCompactionTable(List<Map<String,String>> compactions, long compactionThroughputInBytes, boolean humanReadable, boolean vtableOutput, PrintStream out)
{
if (!compactions.isEmpty())
{
@ -119,9 +119,9 @@ public class CompactionStats extends NodeToolCmd
table.printTo(out);
String remainingTime = "n/a";
if (compactionThroughput != 0)
if (compactionThroughputInBytes != 0)
{
long remainingTimeInSecs = remainingBytes / (1024L * 1024L * compactionThroughput);
long remainingTimeInSecs = remainingBytes / compactionThroughputInBytes;
remainingTime = format("%dh%02dm%02ds", remainingTimeInSecs / 3600, (remainingTimeInSecs % 3600) / 60, (remainingTimeInSecs % 60));
}
out.printf("%25s%10s%n", "Active compaction remaining time : ", remainingTime);

21
src/java/org/apache/cassandra/tools/nodetool/GetCompactionThroughput.java
View File

@ -17,17 +17,34 @@
*/
package org.apache.cassandra.tools.nodetool;
import com.google.common.math.DoubleMath;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "getcompactionthroughput", description = "Print the MiB/s throughput cap for compaction in the system")
@Command(name = "getcompactionthroughput", description = "Print the MiB/s throughput cap for compaction in the system as a rounded number")
public class GetCompactionThroughput extends NodeToolCmd
{
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-d", "--precise-mib" }, description = "Print the MiB/s throughput cap for compaction in the system as a precise number (double)")
private boolean compactionThroughputAsDouble;
@Override
public void execute(NodeProbe probe)
{
probe.output().out.println("Current compaction throughput: " + probe.getCompactionThroughput() + " MiB/s");
double throughput = probe.getCompactionThroughputMebibytesAsDouble();
if (compactionThroughputAsDouble)
probe.output().out.println("Current compaction throughput: " + throughput + " MiB/s");
else
{
if (!DoubleMath.isMathematicalInteger(throughput))
throw new RuntimeException("Use the -d flag to quiet this error and get the exact throughput in MiB/s");
probe.output().out.println("Current compaction throughput: " + probe.getCompactionThroughput() + " MB/s");
}
}
}

60
src/java/org/apache/cassandra/tools/nodetool/GetInterDCStreamThroughput.java
View File

@ -17,27 +17,71 @@
*/
package org.apache.cassandra.tools.nodetool;
import com.google.common.math.DoubleMath;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "getinterdcstreamthroughput", description = "Print the throughput cap for inter-datacenter streaming and entire SSTable inter-datacenter streaming in the system")
@Command(name = "getinterdcstreamthroughput", description = "Print the throughput cap for inter-datacenter streaming and entire SSTable inter-datacenter streaming in the system" +
"in rounded megabits. For precise number, please, use option -d")
public class GetInterDCStreamThroughput extends NodeToolCmd
{
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Print entire SSTable streaming throughput")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Print entire SSTable streaming throughput in MiB/s")
private boolean entireSSTableThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-m", "--mib" }, description = "Print the throughput cap for inter-datacenter streaming in MiB/s")
private boolean interDCStreamThroughputMiB;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-d", "--precise-mbit" }, description = "Print the throughput cap for inter-datacenter streaming in precise Mbits (double)")
private boolean interDCStreamThroughputDoubleMbit;
@Override
public void execute(NodeProbe probe)
{
int throughput = entireSSTableThroughput ? probe.getEntireSSTableInterDCStreamThroughput() : probe.getInterDCStreamThroughput();
int throughput;
double throughputInDouble;
probe.output().out.printf("Current %sinter-datacenter stream throughput: %s%n",
entireSSTableThroughput ? "entire SSTable " : "",
throughput > 0 ? throughput +
(entireSSTableThroughput ? " MiB/s" : " megabits per second")
: "unlimited");
if (entireSSTableThroughput)
{
if (interDCStreamThroughputDoubleMbit || interDCStreamThroughputMiB)
throw new IllegalArgumentException("You cannot use more than one flag with this command");
throughputInDouble = probe.getEntireSSTableInterDCStreamThroughput();
probe.output().out.printf("Current entire SSTable inter-datacenter stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " MiB/s" : "unlimited");
}
else if (interDCStreamThroughputMiB)
{
if (interDCStreamThroughputDoubleMbit)
throw new IllegalArgumentException("You cannot use more than one flag with this command");
throughputInDouble = probe.getInterDCStreamThroughputMibAsDouble();
probe.output().out.printf("Current inter-datacenter stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " MiB/s" : "unlimited");
}
else if (interDCStreamThroughputDoubleMbit)
{
throughputInDouble = probe.getInterDCStreamThroughputAsDouble();
probe.output().out.printf("Current stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " Mb/s" : "unlimited");
}
else
{
throughputInDouble = probe.getInterDCStreamThroughputAsDouble();
throughput = probe.getInterDCStreamThroughput();
if (throughput <= 0)
probe.output().out.printf("Current inter-datacenter stream throughput: unlimited%n");
else if (DoubleMath.isMathematicalInteger(throughputInDouble))
probe.output().out.printf(throughputInDouble + "Current inter-datacenter stream throughput: %s%n", throughput + " Mb/s");
else
throw new RuntimeException("Use the -d flag to quiet this error and get the exact throughput in megabits/s");
}
}
}

59
src/java/org/apache/cassandra/tools/nodetool/GetStreamThroughput.java
View File

@ -17,27 +17,70 @@
*/
package org.apache.cassandra.tools.nodetool;
import com.google.common.math.DoubleMath;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
import org.apache.cassandra.tools.NodeProbe;
import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
@Command(name = "getstreamthroughput", description = "Print the throughput cap for streaming and entire SSTable streaming in the system")
@Command(name = "getstreamthroughput", description = "Print the throughput cap for streaming and entire SSTable streaming in the system in rounded megabits. " +
"For precise number, please, use option -d")
public class GetStreamThroughput extends NodeToolCmd
{
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Print entire SSTable streaming throughput")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Print entire SSTable streaming throughput in MiB/s")
private boolean entireSSTableThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-m", "--mib" }, description = "Print the throughput cap for streaming in MiB/s")
private boolean streamThroughputMiB;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-d", "--precise-mbit" }, description = "Print the throughput cap for streaming in precise Mbits (double)")
private boolean streamThroughputDoubleMbit;
@Override
public void execute(NodeProbe probe)
{
int throughput = entireSSTableThroughput ? probe.getEntireSSTableStreamThroughput() : probe.getStreamThroughput();
int throughput;
double throughputInDouble;
probe.output().out.printf("Current %sstream throughput: %s%n",
entireSSTableThroughput ? "entire SSTable " : "",
throughput > 0 ? throughput +
(entireSSTableThroughput ? " MiB/s" : " megabits per second")
: "unlimited");
if (entireSSTableThroughput)
{
if (streamThroughputDoubleMbit || streamThroughputMiB)
throw new IllegalArgumentException("You cannot use more than one flag with this command");
throughputInDouble = probe.getEntireSSTableStreamThroughput();
probe.output().out.printf("Current entire SSTable stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " MiB/s" : "unlimited");
}
else if (streamThroughputMiB)
{
if (streamThroughputDoubleMbit)
throw new IllegalArgumentException("You cannot use more than one flag with this command");
throughputInDouble = probe.getStreamThroughputMibAsDouble();
probe.output().out.printf("Current stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " MiB/s" : "unlimited");
}
else if (streamThroughputDoubleMbit)
{
throughputInDouble = probe.getStreamThroughputAsDouble();
probe.output().out.printf("Current stream throughput: %s%n",
throughputInDouble > 0 ? throughputInDouble + " Mb/s" : "unlimited");
}
else
{
throughputInDouble = probe.getStreamThroughputAsDouble();
throughput = probe.getStreamThroughput();
if (throughput <= 0)
probe.output().out.printf("Current stream throughput: unlimited%n");
else if (DoubleMath.isMathematicalInteger(throughputInDouble))
probe.output().out.printf("Current stream throughput: %s%n", throughput + " Mb/s");
else
throw new RuntimeException("Use the -d flag to quiet this error and get the exact throughput in megabits/s");
}
}
}

11
src/java/org/apache/cassandra/tools/nodetool/SetInterDCStreamThroughput.java
View File

@ -31,14 +31,23 @@ public class SetInterDCStreamThroughput extends NodeToolCmd
private int interDCStreamThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Set entire SSTable streaming throughput")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Set entire SSTable streaming throughput in MiB/s")
private boolean setEntireSSTableThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-m", "--mib" }, description = "Set streaming throughput in MiB/s")
private boolean interDCStreamThroughputInMebibytes;
@Override
public void execute(NodeProbe probe)
{
if (setEntireSSTableThroughput && interDCStreamThroughputInMebibytes)
throw new IllegalArgumentException("You cannot use -e and -m at the same time");
if (setEntireSSTableThroughput)
probe.setEntireSSTableInterDCStreamThroughput(interDCStreamThroughput);
else if (interDCStreamThroughputInMebibytes )
probe.setInterDCStreamThroughputMiB(interDCStreamThroughput);
else
probe.setInterDCStreamThroughput(interDCStreamThroughput);
}

11
src/java/org/apache/cassandra/tools/nodetool/SetStreamThroughput.java
View File

@ -31,14 +31,23 @@ public class SetStreamThroughput extends NodeToolCmd
private int streamThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Set entire SSTable streaming throughput")
@Option(name = { "-e", "--entire-sstable-throughput" }, description = "Set entire SSTable streaming throughput in MiB/s")
private boolean setEntireSSTableThroughput;
@SuppressWarnings("UnusedDeclaration")
@Option(name = { "-m", "--mib" }, description = "Set streaming throughput in MiB/s")
private boolean streamThroughputInMebibytes;
@Override
public void execute(NodeProbe probe)
{
if (setEntireSSTableThroughput && streamThroughputInMebibytes)
throw new IllegalArgumentException("You cannot use -e and -m at the same time");
if (setEntireSSTableThroughput)
probe.setEntireSSTableStreamThroughput(streamThroughput);
else if (streamThroughputInMebibytes )
probe.setStreamThroughputMiB(streamThroughput);
else
probe.setStreamThroughput(streamThroughput);
}

1
test/conf/cassandra-converters-special-cases-old-names.yaml
View File

@ -4,4 +4,5 @@
#
sstable_preemptive_open_interval_in_mb: -1
index_summary_resize_interval_in_minutes: -1
cache_load_timeout_seconds: -1

1
test/conf/cassandra-converters-special-cases.yaml
View File

@ -4,4 +4,5 @@
#
sstable_preemptive_open_interval:
index_summary_resize_interval:
cache_load_timeout: 0s

138
test/unit/org/apache/cassandra/config/DataRateSpecTest.java
View File

@ -36,52 +36,57 @@ public class DataRateSpecTest
assertEquals(0, new DataRateSpec.LongBytesPerSecondBound("10KiB/s").toMebibytesPerSecond(), 0.1);
assertEquals(10240, new DataRateSpec.LongBytesPerSecondBound("10MiB/s").toKibibytesPerSecond(), 0);
assertEquals(10485760, new DataRateSpec.LongBytesPerSecondBound("10MiB/s").toBytesPerSecond(), 0);
assertEquals(10485760, new DataRateSpec.LongBytesPerSecondBound("10MiB/s").toBytesPerSecond(), 0);
assertEquals(new DataRateSpec.LongBytesPerSecondBound("24MiB/s").toString(), DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(200L).toString());
assertEquals(10, new DataRateSpec.IntMebibytesPerSecondBound("10B/s").toBytesPerSecond(), 0);
assertEquals(10240, new DataRateSpec.IntMebibytesPerSecondBound("10KiB/s").toBytesPerSecond(), 0);
assertEquals(0, new DataRateSpec.IntMebibytesPerSecondBound("10KiB/s").toMebibytesPerSecond(), 0.1);
assertEquals(10240, new DataRateSpec.IntMebibytesPerSecondBound("10MiB/s").toKibibytesPerSecond(), 0);
assertEquals(10485760, new DataRateSpec.IntMebibytesPerSecondBound("10MiB/s").toBytesPerSecond(), 0);
assertEquals(10485760, new DataRateSpec.IntMebibytesPerSecondBound("10MiB/s").toBytesPerSecond(), 0);
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s").toString(), DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(200L).toString());
assertEquals(10, new DataRateSpec.LongBytesPerSecondBound("10MiB/s").toMebibytesPerSecond(), 0);
assertEquals(new DataRateSpec.LongBytesPerSecondBound("25000000B/s").toString(), DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(200L).toString());
}
@Test
public void testOverflowingDuringConversion()
{
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound("2147483649B/s").toBytesPerSecondAsInt(), 0);
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound(2147483649L / 1024L + "KiB/s").toBytesPerSecondAsInt(), 0);
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound(2147483649L / 1024L / 1024 + "MiB/s").toBytesPerSecondAsInt(), 0);
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "B/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: " +
"9223372036854775807B/s. It shouldn't be " +
"more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("9007199254740992KiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: " +
"9007199254740992KiB/s. It shouldn't be " +
"more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("8796093022208MiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: " +
"8796093022208MiB/s. It shouldn't be " +
"more than 9223372036854775806 in bytes_per_second");
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound(2147483646L + "MiB/s").toMegabitsPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE + "MiB/s").toMegabitsPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.LongBytesPerSecondBound(2147483649L + "KiB/s").toKibibytesPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.LongBytesPerSecondBound(2147483649L / 1024L + "MiB/s").toKibibytesPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.LongBytesPerSecondBound(2147483649L + "MiB/s").toMebibytesPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound(2147483647L + "KiB/s").toKibibytesPerSecondAsInt());
assertEquals(Integer.MAX_VALUE, new DataRateSpec.IntMebibytesPerSecondBound(2147483649L / 1024L + "MiB/s").toKibibytesPerSecondAsInt());
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "MiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775807MiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9.223372036854776E18 bytes_per_second. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertEquals(Integer.MAX_VALUE-1, new DataRateSpec.IntMebibytesPerSecondBound(2147483646L + "MiB/s").toMebibytesPerSecondAsInt());
assertThatThrownBy(() -> new DataRateSpec.IntMebibytesPerSecondBound("2147483648MiB/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2147483648MiB/s. " +
"It shouldn't be more than 2147483646 in mebibytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.IntMebibytesPerSecondBound(2147483648L)).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2.147483648E9 mebibytes_per_second. " +
"It shouldn't be more than 2147483646 in mebibytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.IntMebibytesPerSecondBound((Integer.MAX_VALUE * 1024L + 1L) + "KiB/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2199023254529KiB/s. " +
"It shouldn't be more than 2147483646 in mebibytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.IntMebibytesPerSecondBound((Integer.MAX_VALUE * 1024L * 1024 + 1L) + "B/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2251799812636673B/s. " +
"It shouldn't be more than 2147483646 in mebibytes_per_second");
assertThatThrownBy(() -> DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(2147483648L)).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2147483648 megabits per second; " +
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("9007199254740992KiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9007199254740992KiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("8796093022208MiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 8796093022208MiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(Integer.MAX_VALUE))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2147483647 megabits per second; " +
"stream_throughput_outbound and " +
"inter_dc_stream_throughput_outbound should " +
"be between 0 and 2147483647 in megabits per second");
"be between 0 and 2147483646 in megabits per second");
}
@Test
@ -102,8 +107,9 @@ public class DataRateSpecTest
.hasMessageContaining("Invalid data rate: 10");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("-10b/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: -10b/s");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(-10, DataRateSpec.DataRateUnit.BYTES_PER_SECOND)).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: value must be non-negative");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(-10, DataRateSpec.DataRateUnit.BYTES_PER_SECOND))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: value must be non-negative");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("10xb/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 10xb/s");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("9223372036854775809B/s")
@ -115,10 +121,6 @@ public class DataRateSpecTest
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound("9223372036854775809MiB/s")
.toBytesPerSecond()).isInstanceOf(NumberFormatException.class)
.hasMessageContaining("For input string: \"9223372036854775809\"");
assertThatThrownBy(() -> new DataRateSpec.IntMebibytesPerSecondBound("2147483648MiB/s")
.toBytesPerSecond()).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 2147483648MiB/s. It shouldn't be more" +
" than 2147483646 in mebibytes_per_second");
}
@Test
@ -127,25 +129,27 @@ public class DataRateSpecTest
//just test the cast to Int as currently we don't even have any long bound rates and there is a very low probability of ever having them
assertEquals(Integer.MAX_VALUE, new DataRateSpec.LongBytesPerSecondBound("92233720368547758B/s").toBytesPerSecondAsInt());
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "B/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775807B/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "MiB/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775807MiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE - 5 + "KiB/s")).isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775802KiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "B/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775807B/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE + "MiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775807MiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
assertThatThrownBy(() -> new DataRateSpec.LongBytesPerSecondBound(Long.MAX_VALUE - 5 + "KiB/s"))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid data rate: 9223372036854775802KiB/s. " +
"It shouldn't be more than 9223372036854775806 in bytes_per_second");
}
@Test
public void testValidUnits()
{
// we need toString as internally it is double and they are not 0.0 equal but for the end user the double numbers don't exist
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("24MiB/s").toString(), DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(200).toString());
assertEquals(new DataRateSpec.LongBytesPerSecondBound("25000000B/s"), DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(200));
assertEquals(new DataRateSpec.LongBytesPerSecondBound("24MiB/s"), new DataRateSpec.LongBytesPerSecondBound("25165824B/s"));
}
@SuppressWarnings("AssertBetweenInconvertibleTypes")
@Test
public void testEquals()
{
@ -153,26 +157,13 @@ public class DataRateSpecTest
assertEquals(new DataRateSpec.LongBytesPerSecondBound("10KiB/s"), new DataRateSpec.LongBytesPerSecondBound("10240B/s"));
assertEquals(new DataRateSpec.LongBytesPerSecondBound("10240B/s"), new DataRateSpec.LongBytesPerSecondBound("10KiB/s"));
assertNotEquals(new DataRateSpec.LongBytesPerSecondBound("0KiB/s"), new DataRateSpec.LongBytesPerSecondBound("10MiB/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10B/s"), new DataRateSpec.IntMebibytesPerSecondBound("10B/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10KiB/s"), new DataRateSpec.IntMebibytesPerSecondBound("10240B/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10240B/s"), new DataRateSpec.IntMebibytesPerSecondBound("10KiB/s"));
assertNotEquals(new DataRateSpec.IntMebibytesPerSecondBound("0KiB/s"), new DataRateSpec.IntMebibytesPerSecondBound("10MiB/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10B/s"), new DataRateSpec.LongBytesPerSecondBound("10B/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10KiB/s"), new DataRateSpec.LongBytesPerSecondBound("10240B/s"));
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound("10240B/s"), new DataRateSpec.LongBytesPerSecondBound("10KiB/s"));
assertNotEquals(new DataRateSpec.IntMebibytesPerSecondBound("0KiB/s"), new DataRateSpec.LongBytesPerSecondBound("10MiB/s"));
}
@Test
public void thereAndBackLongBytesPerSecondBound()
{
Gen<DataRateSpec.DataRateUnit> unitGen = SourceDSL.arbitrary().enumValues(DataRateSpec.DataRateUnit.class);
// DataRateSpec is a special case where we have double so we can accomodate the backward compatibility for parameters which were in
// megabits per second before without losing precision
// Extremely big numbers might be not completely accurate, that is why here Long.MAX_VALUE is not failing for bytes being >= Long.MAX_VALUE
Gen<Long> valueGen = SourceDSL.longs().between(0, Long.MAX_VALUE/1024L/1024L); // the biggest value in MiB/s that won't lead to B/s overflow
Gen<Long> valueGen = SourceDSL.longs().between(0, 8796093022207L); // the biggest value in MiB/s that won't lead to B/s overflow
qt().forAll(valueGen, unitGen).check((value, unit) -> {
DataRateSpec.LongBytesPerSecondBound there = new DataRateSpec.LongBytesPerSecondBound(value, unit);
DataRateSpec.LongBytesPerSecondBound back = new DataRateSpec.LongBytesPerSecondBound(there.toString());
@ -181,15 +172,10 @@ public class DataRateSpecTest
}
@Test
public void thereAndBackIntMebibytesPerSecondBound()
public void testToString()
{
Gen<DataRateSpec.DataRateUnit> unitGen = SourceDSL.arbitrary().enumValues(DataRateSpec.DataRateUnit.class);
Gen<Long> valueGen = SourceDSL.longs().between(0, Integer.MAX_VALUE-1); // max MiB/s
qt().forAll(valueGen, unitGen).check((value, unit) -> {
DataRateSpec.IntMebibytesPerSecondBound there = new DataRateSpec.IntMebibytesPerSecondBound(value, unit);
DataRateSpec.IntMebibytesPerSecondBound back = new DataRateSpec.IntMebibytesPerSecondBound(there.toString());
return there.equals(back) && back.equals(there);
});
DataRateSpec testProperty = new DataRateSpec.LongBytesPerSecondBound("5B/s");
assertEquals("5B/s", testProperty.toString());
}
@Test

2
test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
View File

@ -106,7 +106,7 @@ public class DatabaseDescriptorRefTest
"org.apache.cassandra.config.DataStorageSpec$LongBytesBound",
"org.apache.cassandra.config.DataStorageSpec$LongMebibytesBound",
"org.apache.cassandra.config.DurationSpec",
"org.apache.cassandra.config.DataRateSpec$IntMebibytesPerSecondBound",
"org.apache.cassandra.config.DataRateSpec$LongBytesPerSecondBound",
"org.apache.cassandra.config.DurationSpec$LongMillisecondsBound",
"org.apache.cassandra.config.DurationSpec$LongNanosecondsBound",
"org.apache.cassandra.config.DurationSpec$LongSecondsBound",

38
test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
View File

@ -586,6 +586,44 @@ public class DatabaseDescriptorTest
}
}
@Test
public void testUpperBoundStreamingConfigOnStartup()
{
Config config = DatabaseDescriptor.loadConfig();
String expectedMsg = "Invalid value of entire_sstable_stream_throughput_outbound:";
config.entire_sstable_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND);
validateProperty(expectedMsg);
expectedMsg = "Invalid value of entire_sstable_stream_throughput_outbound:";
config.entire_sstable_inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND);
validateProperty(expectedMsg);
expectedMsg = "Invalid value of stream_throughput_outbound:";
config.stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE * 125_000L);
validateProperty(expectedMsg);
expectedMsg = "Invalid value of inter_dc_stream_throughput_outbound:";
config.inter_dc_stream_throughput_outbound = new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE * 125_000L);
validateProperty(expectedMsg);
expectedMsg = "compaction_throughput:";
config.compaction_throughput = new DataRateSpec.LongBytesPerSecondBound(Integer.MAX_VALUE, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND);
validateProperty(expectedMsg);
}
private static void validateProperty(String expectedMsg)
{
try
{
DatabaseDescriptor.validateUpperBoundStreamingConfig();
}
catch (ConfigurationException ex)
{
Assert.assertEquals(expectedMsg, ex.getMessage());
}
}
@Test
public void testApplyTokensConfigInitialTokensNotSetNumTokensSet()
{

8
test/unit/org/apache/cassandra/config/LoadOldYAMLBackwardCompatibilityTest.java
View File

@ -72,11 +72,11 @@ public class LoadOldYAMLBackwardCompatibilityTest
assertEquals(new DataStorageSpec.IntKibibytesBound(4), config.column_index_size);
assertEquals(new DataStorageSpec.IntKibibytesBound(2), config.column_index_cache_size);
assertEquals(new DataStorageSpec.IntKibibytesBound(5), config.batch_size_warn_threshold);
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(64), config.compaction_throughput);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(64, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND), config.compaction_throughput);
assertEquals(new DataStorageSpec.IntMebibytesBound(50), config.min_free_space_per_drive);
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(23841858).toString(), config.stream_throughput_outbound.toString());
assertEquals(DataRateSpec.IntMebibytesPerSecondBound.megabitsPerSecondInMebibytesPerSecond(200000000).toString(), config.stream_throughput_outbound.toString());
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(24), config.inter_dc_stream_throughput_outbound);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(25000000000000L).toString(), config.stream_throughput_outbound.toString());
assertEquals(DataRateSpec.LongBytesPerSecondBound.megabitsPerSecondInBytesPerSecond(200000000), config.stream_throughput_outbound);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(24L * 1024L * 1024L), config.inter_dc_stream_throughput_outbound);
assertNull(config.commitlog_total_space);
assertEquals(new DurationSpec.IntMillisecondsBound(0.0, TimeUnit.MILLISECONDS), config.commitlog_sync_group_window);
assertEquals(new DurationSpec.IntMillisecondsBound(0), config.commitlog_sync_period);

6
test/unit/org/apache/cassandra/config/ParseAndConvertUnitsTest.java
View File

@ -112,8 +112,8 @@ public class ParseAndConvertUnitsTest
assertEquals(new DataStorageSpec.IntBytesBound(1, MEBIBYTES), config.native_transport_receive_queue_capacity);
//Confirm rate parameters were successfully parsed with the default values in cassandra.yaml
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(0), config.compaction_throughput);
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(23841858), config.stream_throughput_outbound);
assertEquals(new DataRateSpec.IntMebibytesPerSecondBound(24), config.inter_dc_stream_throughput_outbound);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(0), config.compaction_throughput);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(23841858, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND), config.stream_throughput_outbound);
assertEquals(new DataRateSpec.LongBytesPerSecondBound(24, DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND), config.inter_dc_stream_throughput_outbound);
}
}

12
test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
View File

@ -117,10 +117,12 @@ public class YamlConfigurationLoaderTest
Config c = load("test/conf/cassandra-converters-special-cases.yaml");
assertThat(c.sstable_preemptive_open_interval).isNull();
assertThat(c.index_summary_resize_interval).isNull();
assertThat(c.cache_load_timeout).isEqualTo(new DurationSpec.IntSecondsBound("0s"));
c = load("test/conf/cassandra-converters-special-cases-old-names.yaml");
assertThat(c.sstable_preemptive_open_interval).isNull();
assertThat(c.index_summary_resize_interval).isNull();
assertThat(c.cache_load_timeout).isEqualTo(new DurationSpec.IntSecondsBound("0s"));
}
@Test
@ -131,17 +133,19 @@ public class YamlConfigurationLoaderTest
map.put("index_summary_resize_interval", null);
Config c = YamlConfigurationLoader.fromMap(map, true, Config.class);
assert c.sstable_preemptive_open_interval == null;
assert c.index_summary_resize_interval == null;
assertThat(c.sstable_preemptive_open_interval).isNull();
assertThat(c.index_summary_resize_interval).isNull();
map = ImmutableMap.of(
"sstable_preemptive_open_interval_in_mb", "-1",
"index_summary_resize_interval_in_minutes", "-1"
"index_summary_resize_interval_in_minutes", "-1",
"cache_load_timeout_seconds", "-1"
);
c = YamlConfigurationLoader.fromMap(map, Config.class);
assertThat(c.sstable_preemptive_open_interval).isNull();
assertThat(c.index_summary_resize_interval).isNull();
assertThat(c.cache_load_timeout).isEqualTo(new DurationSpec.IntSecondsBound("0s"));
}
@Test
@ -340,7 +344,7 @@ public class YamlConfigurationLoaderTest
.hasRootCauseInstanceOf(IllegalArgumentException.class)
.hasRootCauseMessage("Invalid data rate: value must be non-negative");
// MEGABITS_TO_MEBIBYTES_PER_SECOND_DATA_RATE
// MEGABITS_TO_BYTES_PER_SECOND_DATA_RATE
assertThat(from("stream_throughput_outbound_megabits_per_sec", "42").stream_throughput_outbound.toMegabitsPerSecondAsInt()).isEqualTo(42);
assertThatThrownBy(() -> from("stream_throughput_outbound_megabits_per_sec", -2).stream_throughput_outbound.toMegabitsPerSecondAsInt())
.hasRootCauseInstanceOf(IllegalArgumentException.class)

4
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -74,6 +74,7 @@ public class KeyCacheTest
private static final String COLUMN_FAMILY7 = "Standard7";
private static final String COLUMN_FAMILY8 = "Standard8";
private static final String COLUMN_FAMILY9 = "Standard9";
private static final String COLUMN_FAMILY10 = "Standard10";
private static final String COLUMN_FAMILY_K2_1 = "Standard1";
@ -92,7 +93,8 @@ public class KeyCacheTest
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY6),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY7),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY8),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY9));
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY9),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY10));
SchemaLoader.createKeyspace(KEYSPACE2,
KeyspaceParams.simple(1),

10
test/unit/org/apache/cassandra/db/virtual/SettingsTableTest.java
View File

@ -31,6 +31,7 @@ import org.junit.Test;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.cql3.CQLTester;
@ -58,6 +59,7 @@ public class SettingsTableTest extends CQLTester
config.server_encryption_options.applyConfig();
config.sstable_preemptive_open_interval = null;
config.index_summary_resize_interval = null;
config.cache_load_timeout = new DurationSpec.IntSecondsBound(0);
table = new SettingsTable(KS_NAME, config);
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KS_NAME, ImmutableList.of(table)));
disablePreparedReuseForTest();
@ -117,11 +119,17 @@ public class SettingsTableTest extends CQLTester
q = "SELECT * FROM vts.settings WHERE name = 'sstable_preemptive_open_interval_in_mb';";
assertRowsNet(executeNet(q), new Object[] {"sstable_preemptive_open_interval_in_mb", "-1"});
// test MINUTES_CUSTOM_DURATION - test to be added later whe
// test MINUTES_CUSTOM_DURATION converter
q = "SELECT * FROM vts.settings WHERE name = 'index_summary_resize_interval';";
assertRowsNet(executeNet(q), new Object[] {"index_summary_resize_interval", null});
q = "SELECT * FROM vts.settings WHERE name = 'index_summary_resize_interval_in_minutes';";
assertRowsNet(executeNet(q), new Object[] {"index_summary_resize_interval_in_minutes", "-1"});
// test NEGATIVE_SECONDS_DURATION converter
q = "SELECT * FROM vts.settings WHERE name = 'cache_load_timeout';";
assertRowsNet(executeNet(q), new Object[] {"cache_load_timeout", "0s"});
q = "SELECT * FROM vts.settings WHERE name = 'cache_load_timeout_seconds';";
assertRowsNet(executeNet(q), new Object[] {"cache_load_timeout_seconds", "0"});
}
private String getValue(Property prop)

44
test/unit/org/apache/cassandra/streaming/StreamManagerTest.java
View File

@ -32,27 +32,27 @@ import static org.junit.Assert.assertEquals;
public class StreamManagerTest
{
private static double defaultStreamThroughputMebibytesPerSec;
private static double defaultInterDCStreamThroughputMebibytesPerSec;
private static double defaultStreamThroughputBytesPerSec;
private static double defaultInterDCStreamThroughputBytesPerSec;
private static final int MAX_INT_CONFIG_VALUE = Integer.MAX_VALUE - 1;
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES = DataRateSpec.IntMebibytesPerSecondBound
.megabitsPerSecondInMebibytesPerSecond(MAX_INT_CONFIG_VALUE)
.toMebibytesPerSecond();
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_BYTES = DataRateSpec.LongBytesPerSecondBound
.megabitsPerSecondInBytesPerSecond(MAX_INT_CONFIG_VALUE)
.toBytesPerSecond();
private static double defaultEntireSSTableStreamThroughputMebibytesPerSec;
private static double defaultEntireSSTableInterDCStreamThroughputMebibytesPerSec;
private static double defaultEntireSSTableStreamThroughputBytesPerSec;
private static double defaultEntireSSTableInterDCStreamThroughputBytesPerSec;
private static final double MEBIBYTES_PER_MEGABIT = 0.119209289550781;
private static final double BYTES_PER_MEGABIT = 125_000;
@BeforeClass
public static void setupClass()
{
Config c = DatabaseDescriptor.loadConfig();
defaultStreamThroughputMebibytesPerSec = c.stream_throughput_outbound.toMebibytesPerSecond();
defaultInterDCStreamThroughputMebibytesPerSec = c.inter_dc_stream_throughput_outbound.toMebibytesPerSecond();
defaultEntireSSTableStreamThroughputMebibytesPerSec = c.entire_sstable_stream_throughput_outbound.toMebibytesPerSecond();
defaultEntireSSTableInterDCStreamThroughputMebibytesPerSec = c.entire_sstable_inter_dc_stream_throughput_outbound.toMebibytesPerSecond();
defaultStreamThroughputBytesPerSec = c.stream_throughput_outbound.toBytesPerSecond();
defaultInterDCStreamThroughputBytesPerSec = c.inter_dc_stream_throughput_outbound.toBytesPerSecond();
defaultEntireSSTableStreamThroughputBytesPerSec = c.entire_sstable_stream_throughput_outbound.toBytesPerSecond();
defaultEntireSSTableInterDCStreamThroughputBytesPerSec = c.entire_sstable_inter_dc_stream_throughput_outbound.toBytesPerSecond();
DatabaseDescriptor.daemonInitialization(() -> c);
}
@ -61,15 +61,15 @@ public class StreamManagerTest
public void testUpdateStreamThroughput()
{
// Initialized value check
assertEquals(defaultStreamThroughputMebibytesPerSec * BYTES_PER_MEBIBYTE, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
assertEquals(defaultStreamThroughputBytesPerSec, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setStreamThroughputMbitPerSec(500); //60MiB/s
assertEquals(500 * MEBIBYTES_PER_MEGABIT * BYTES_PER_MEBIBYTE, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
assertEquals(500 * BYTES_PER_MEGABIT, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Max positive value check
StorageService.instance.setStreamThroughputMbitPerSec(MAX_INT_CONFIG_VALUE);
assertEquals(INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES * BYTES_PER_MEBIBYTE, StreamRateLimiter.getRateLimiterRateInBytes(), 0.04);
assertEquals(INTEGER_MAX_VALUE_MEGABITS_IN_BYTES, StreamRateLimiter.getRateLimiterRateInBytes(), 0);
// Zero value check
StorageService.instance.setStreamThroughputMbitPerSec(0);
@ -80,11 +80,11 @@ public class StreamManagerTest
public void testUpdateEntireSSTableStreamThroughput()
{
// Initialized value check (defaults to StreamRateLimiter.getRateLimiterRateInBytes())
assertEquals(defaultEntireSSTableStreamThroughputMebibytesPerSec * BYTES_PER_MEBIBYTE, StreamRateLimiter.getEntireSSTableRateLimiterRateInBytes(), 0);
assertEquals(defaultEntireSSTableStreamThroughputBytesPerSec, StreamRateLimiter.getEntireSSTableRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setEntireSSTableStreamThroughputMebibytesPerSec(1500);
assertEquals(1500d * BYTES_PER_MEBIBYTE, Math.round(StreamRateLimiter.getEntireSSTableRateLimiterRateInBytes()), 0);
assertEquals(1500d * BYTES_PER_MEBIBYTE, StreamRateLimiter.getEntireSSTableRateLimiterRateInBytes(), 0);
// Max positive value check
StorageService.instance.setEntireSSTableStreamThroughputMebibytesPerSec(MAX_INT_CONFIG_VALUE);
@ -99,15 +99,15 @@ public class StreamManagerTest
public void testUpdateInterDCStreamThroughput()
{
// Initialized value check
assertEquals(defaultInterDCStreamThroughputMebibytesPerSec * BYTES_PER_MEBIBYTE, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
assertEquals(defaultInterDCStreamThroughputBytesPerSec, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setInterDCStreamThroughputMbitPerSec(200); //approximately 24MiB/s
assertEquals(200 * MEBIBYTES_PER_MEGABIT * BYTES_PER_MEBIBYTE, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
assertEquals(200 * BYTES_PER_MEGABIT, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Max positive value check
StorageService.instance.setInterDCStreamThroughputMbitPerSec(MAX_INT_CONFIG_VALUE);
assertEquals(INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES * BYTES_PER_MEBIBYTE, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0.04);
assertEquals(INTEGER_MAX_VALUE_MEGABITS_IN_BYTES, StreamRateLimiter.getInterDCRateLimiterRateInBytes(), 0);
// Zero value check
StorageService.instance.setInterDCStreamThroughputMbitPerSec(0);
@ -118,7 +118,7 @@ public class StreamManagerTest
public void testUpdateEntireSSTableInterDCStreamThroughput()
{
// Initialized value check (Defaults to StreamRateLimiter.getInterDCRateLimiterRateInBytes())
assertEquals(defaultEntireSSTableInterDCStreamThroughputMebibytesPerSec * BYTES_PER_MEBIBYTE, StreamRateLimiter.getEntireSSTableInterDCRateLimiterRateInBytes(), 0);
assertEquals(defaultEntireSSTableInterDCStreamThroughputBytesPerSec, StreamRateLimiter.getEntireSSTableInterDCRateLimiterRateInBytes(), 0);
// Positive value check
StorageService.instance.setEntireSSTableInterDCStreamThroughputMebibytesPerSec(1200);
@ -126,7 +126,7 @@ public class StreamManagerTest
// Max positive value check
StorageService.instance.setEntireSSTableInterDCStreamThroughputMebibytesPerSec(MAX_INT_CONFIG_VALUE);
assertEquals((MAX_INT_CONFIG_VALUE) * BYTES_PER_MEBIBYTE, StreamRateLimiter.getEntireSSTableInterDCRateLimiterRateInBytes(), 0);
assertEquals(MAX_INT_CONFIG_VALUE * BYTES_PER_MEBIBYTE, StreamRateLimiter.getEntireSSTableInterDCRateLimiterRateInBytes(), 0);
// Zero value check
StorageService.instance.setEntireSSTableInterDCStreamThroughputMebibytesPerSec(0);

51
test/unit/org/apache/cassandra/tools/nodetool/SetGetCompactionThroughputTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.tools.nodetool;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import static org.apache.cassandra.tools.ToolRunner.ToolResult;
@ -32,6 +33,8 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
public class SetGetCompactionThroughputTest extends CQLTester
{
private static final int MAX_INT_CONFIG_VALUE_IN_MBIT = Integer.MAX_VALUE - 1;
@BeforeClass
public static void setup() throws Exception
{
@ -53,7 +56,13 @@ public class SetGetCompactionThroughputTest extends CQLTester
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(Integer.MAX_VALUE - 1);
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE_IN_MBIT);
}
@Test
public void testUpperBound()
{
assertSetInvalidThroughputMib(String.valueOf(Integer.MAX_VALUE));
}
@Test
@ -67,6 +76,8 @@ public class SetGetCompactionThroughputTest extends CQLTester
{
assertSetInvalidThroughput("1.2", "compaction_throughput: can not convert \"1.2\" to a Integer");
assertSetInvalidThroughput("value", "compaction_throughput: can not convert \"value\" to a Integer");
assertSetInvalidThroughput();
assertPreciseMibFlagNeeded();
}
private static void assertSetGetValidThroughput(int throughput)
@ -76,6 +87,7 @@ public class SetGetCompactionThroughputTest extends CQLTester
assertThat(tool.getStdout()).isEmpty();
assertGetThroughput(throughput);
assertGetThroughputDouble(throughput);
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
@ -86,14 +98,49 @@ public class SetGetCompactionThroughputTest extends CQLTester
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertSetInvalidThroughput()
{
DatabaseDescriptor.setCompactionThroughputBytesPerSec(500);
ToolResult tool = invokeNodetool("getstreamthroughput");
assertThat(tool.getExitCode()).isEqualTo(2);
assertThat(tool.getStderr()).contains("Use the -d flag to quiet this error and get the exact throughput in megabits/s");
}
private static void assertSetInvalidThroughputMib(String throughput)
{
ToolResult tool = invokeNodetool("setcompactionthroughput", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("compaction_throughput: 2147483647 is too large; it should be less than" +
" 2147483647 in MiB/s");
}
private static void assertPreciseMibFlagNeeded()
{
DatabaseDescriptor.setCompactionThroughputBytesPerSec(15);
ToolResult tool = invokeNodetool("getcompactionthroughput");
assertThat(tool.getExitCode()).isEqualTo(2);
assertThat(tool.getStderr()).contains("Use the -d flag to quiet this error and get the exact throughput in MiB/s");
}
private static void assertGetThroughput(int expected)
{
ToolResult tool = invokeNodetool("getcompactionthroughput");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current compaction throughput: " + expected + " MB/s");
else
assertThat(tool.getStdout()).contains("Current compaction throughput: 0 MB/s");
}
private static void assertGetThroughputDouble(double expected)
{
ToolResult tool = invokeNodetool("getcompactionthroughput", "-d");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current compaction throughput: " + expected + " MiB/s");
else
assertThat(tool.getStdout()).contains("Current compaction throughput: 0 MiB/s");
assertThat(tool.getStdout()).contains("Current compaction throughput: 0.0 MiB/s");
}
}

20
test/unit/org/apache/cassandra/tools/nodetool/SetGetEntireSSTableInterDCStreamThroughputTest.java
View File

@ -34,6 +34,8 @@ import static org.assertj.core.api.Assertions.withPrecision;
*/
public class SetGetEntireSSTableInterDCStreamThroughputTest extends CQLTester
{
private static final int MAX_INT_CONFIG_VALUE_MIB = Integer.MAX_VALUE - 1;
@BeforeClass
public static void setup() throws Exception
{
@ -55,7 +57,13 @@ public class SetGetEntireSSTableInterDCStreamThroughputTest extends CQLTester
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(Integer.MAX_VALUE - 1, (Integer.MAX_VALUE - 1) * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE_MIB, MAX_INT_CONFIG_VALUE_MIB * StreamRateLimiter.BYTES_PER_MEBIBYTE);
}
@Test
public void testUpperBound()
{
assertSetInvalidEntireSStableInterDCThroughputMib(String.valueOf(Integer.MAX_VALUE));
}
@Test
@ -90,7 +98,15 @@ public class SetGetEntireSSTableInterDCStreamThroughputTest extends CQLTester
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertGetThroughput(int expected)
private static void assertSetInvalidEntireSStableInterDCThroughputMib(String throughput)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", "-e", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("entire_sstable_inter_dc_stream_throughput_outbound: 2147483647 is too large;" +
" it should be less than 2147483647 in MiB/s");
}
private static void assertGetThroughput(double expected)
{
ToolResult tool = invokeNodetool("getinterdcstreamthroughput", "-e");
tool.assertOnCleanExit();

20
test/unit/org/apache/cassandra/tools/nodetool/SetGetEntireSSTableStreamThroughputTest.java
View File

@ -34,6 +34,8 @@ import static org.assertj.core.api.Assertions.withPrecision;
*/
public class SetGetEntireSSTableStreamThroughputTest extends CQLTester
{
private static final int MAX_INT_CONFIG_VALUE_MIB = Integer.MAX_VALUE - 1;
@BeforeClass
public static void setup() throws Exception
{
@ -55,7 +57,13 @@ public class SetGetEntireSSTableStreamThroughputTest extends CQLTester
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(Integer.MAX_VALUE - 1, (Integer.MAX_VALUE - 1) * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE_MIB, MAX_INT_CONFIG_VALUE_MIB * StreamRateLimiter.BYTES_PER_MEBIBYTE);
}
@Test
public void testUpperBound()
{
assertSetInvalidEntireSStableThroughputMib(String.valueOf(Integer.MAX_VALUE));
}
@Test
@ -87,6 +95,14 @@ public class SetGetEntireSSTableStreamThroughputTest extends CQLTester
assertThat(StreamRateLimiter.getEntireSSTableRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetInvalidEntireSStableThroughputMib(String throughput)
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-e", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("entire_sstable_stream_throughput_outbound: 2147483647 is too large; it " +
"should be less than 2147483647 in MiB/s");
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
{
ToolResult tool = throughput == null ? invokeNodetool("setstreamthroughput", "-e")
@ -95,7 +111,7 @@ public class SetGetEntireSSTableStreamThroughputTest extends CQLTester
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertGetThroughput(int expected)
private static void assertGetThroughput(double expected)
{
ToolResult tool = invokeNodetool("getstreamthroughput", "-e");
tool.assertOnCleanExit();

120
test/unit/org/apache/cassandra/tools/nodetool/SetGetInterDCStreamThroughputTest.java
View File

@ -35,10 +35,13 @@ import static org.assertj.core.api.Assertions.withPrecision;
*/
public class SetGetInterDCStreamThroughputTest extends CQLTester
{
private static final int MAX_INT_CONFIG_VALUE = Integer.MAX_VALUE - 1;
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES = DataRateSpec.IntMebibytesPerSecondBound
.megabitsPerSecondInMebibytesPerSecond(MAX_INT_CONFIG_VALUE)
.toMebibytesPerSecond();
private static final int MAX_INT_CONFIG_VALUE_IN_MBIT = Integer.MAX_VALUE - 1;
private static final double BYTES_PER_MEGABIT = 125_000;
private static final int MAX_INT_CONFIG_VALUE_MIB = (int) (MAX_INT_CONFIG_VALUE_IN_MBIT * BYTES_PER_MEGABIT) / 1024 / 1024;
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_BYTES = DataRateSpec.LongBytesPerSecondBound
.megabitsPerSecondInBytesPerSecond(MAX_INT_CONFIG_VALUE_IN_MBIT)
.toBytesPerSecond();
private static final double MEBIBYTES_PER_MEGABIT = 0.11920928955078125;
@BeforeClass
public static void setup() throws Exception
@ -55,7 +58,9 @@ public class SetGetInterDCStreamThroughputTest extends CQLTester
@Test
public void testPositive()
{
assertSetGetValidThroughput(7, 0.834465026855467 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(7, 7 * BYTES_PER_MEGABIT);
assertSetGetValidThroughputMiB(7, 7 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(7, 7 * BYTES_PER_MEGABIT);
}
@Test
@ -64,19 +69,32 @@ public class SetGetInterDCStreamThroughputTest extends CQLTester
// As part of CASSANDRA-15234 we had to do some tweaks with precision. This test has to ensure no regressions
// happen, hopefully. Internally data rate parameters values and rate limitter are set in double. Users can set
// and get only integers
assertSetGetValidThroughput(1, 0.119209289550781 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(1, 1 * BYTES_PER_MEGABIT);
assertSetGetValidThroughputMiB(1, 1 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(1, 1 * BYTES_PER_MEGABIT);
}
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE, INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE_IN_MBIT, INTEGER_MAX_VALUE_MEGABITS_IN_BYTES);
assertSetGetValidThroughputMiB(MAX_INT_CONFIG_VALUE_MIB, MAX_INT_CONFIG_VALUE_MIB * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(MAX_INT_CONFIG_VALUE_IN_MBIT, INTEGER_MAX_VALUE_MEGABITS_IN_BYTES);
}
@Test
public void testUpperBound()
{
assertSetInvalidThroughputMib(String.valueOf(Integer.MAX_VALUE));
assertSetInvalidThroughputMbit(String.valueOf(Integer.MAX_VALUE));
}
@Test
public void testZero()
{
assertSetGetValidThroughput(0, Double.MAX_VALUE);
assertSetGetValidThroughputMiB(0, Double.MAX_VALUE);
assertSetMbitGetMibValidThroughput(0, Double.MAX_VALUE);
}
@Test
@ -84,6 +102,8 @@ public class SetGetInterDCStreamThroughputTest extends CQLTester
{
assertSetInvalidThroughput("1.2", "inter_dc_stream_throughput: can not convert \"1.2\" to a int");
assertSetInvalidThroughput("value", "inter_dc_stream_throughput: can not convert \"value\" to a int");
assertSetGetMoreFlagsIsInvalid();
assertDFlagNeeded();
}
private static void assertSetGetValidThroughput(int throughput, double rateInBytes)
@ -97,6 +117,39 @@ public class SetGetInterDCStreamThroughputTest extends CQLTester
assertThat(StreamRateLimiter.getInterDCRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.04));
}
private static void assertDFlagNeeded()
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-m", String.valueOf(1));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
tool = invokeNodetool("getstreamthroughput");
assertThat(tool.getExitCode()).isEqualTo(2);
assertThat(tool.getStderr()).contains("Use the -d flag to quiet this error and get the exact throughput in megabits/s");
}
private static void assertSetGetValidThroughputMiB(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", "-m", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughputMiB(throughput);
assertThat(StreamRateLimiter.getInterDCRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetMbitGetMibValidThroughput(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughputMiB(throughput * MEBIBYTES_PER_MEGABIT);
assertThat(StreamRateLimiter.getInterDCRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
{
ToolResult tool = throughput == null ? invokeNodetool("setinterdcstreamthroughput")
@ -105,13 +158,64 @@ public class SetGetInterDCStreamThroughputTest extends CQLTester
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertSetInvalidThroughputMib(String throughput)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", "-m", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("inter_dc_stream_throughput_outbound: 2147483647 is too large; it should be" +
" less than 2147483647 in megabits/s");
}
private static void assertSetInvalidThroughputMbit(String throughput)
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("Invalid data rate: 2147483647 megabits per second; stream_throughput_outbound" +
" and inter_dc_stream_throughput_outbound should be between 0 and 2147483646 in " +
"megabits per second");
}
private static void assertSetGetMoreFlagsIsInvalid()
{
ToolResult tool = invokeNodetool("setinterdcstreamthroughput", "-m", "5", "-e", "5");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use -e and -m at the same time");
tool = invokeNodetool("getinterdcstreamthroughput", "-m", "-e");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getinterdcstreamthroughput", "-m", "-d");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getinterdcstreamthroughput", "-d", "-e");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getinterdcstreamthroughput", "-m", "-e", "-d");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
}
private static void assertGetThroughput(int expected)
{
ToolResult tool = invokeNodetool("getinterdcstreamthroughput");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: " + expected + " megabits per second");
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: " + expected + " Mb/s");
else
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: unlimited");
}
private static void assertGetThroughputMiB(double expected)
{
ToolResult tool = invokeNodetool("getinterdcstreamthroughput", "-m");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: " + expected + " MiB/s");
else
assertThat(tool.getStdout()).contains("Current inter-datacenter stream throughput: unlimited");
}

120
test/unit/org/apache/cassandra/tools/nodetool/SetGetStreamThroughputTest.java
View File

@ -36,10 +36,13 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
public class SetGetStreamThroughputTest extends CQLTester
{
private static final int MAX_INT_CONFIG_VALUE = Integer.MAX_VALUE - 1;
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES = DataRateSpec.IntMebibytesPerSecondBound
.megabitsPerSecondInMebibytesPerSecond(MAX_INT_CONFIG_VALUE)
.toMebibytesPerSecond();
private static final int MAX_INT_CONFIG_VALUE_IN_MBIT = Integer.MAX_VALUE - 1;
private static final double BYTES_PER_MEGABIT = 125_000;
private static final int MAX_INT_CONFIG_VALUE_MIB = (int) (MAX_INT_CONFIG_VALUE_IN_MBIT * BYTES_PER_MEGABIT) / 1024 / 1024;
private static final double INTEGER_MAX_VALUE_MEGABITS_IN_BYTES = DataRateSpec.LongBytesPerSecondBound
.megabitsPerSecondInBytesPerSecond(MAX_INT_CONFIG_VALUE_IN_MBIT)
.toBytesPerSecond();
private static final double MEBIBYTES_PER_MEGABIT = 0.11920928955078125;
@BeforeClass
public static void setup() throws Exception
@ -56,7 +59,9 @@ public class SetGetStreamThroughputTest extends CQLTester
@Test
public void testPositive()
{
assertSetGetValidThroughput(7, 0.834465026855467 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(7, 7 * BYTES_PER_MEGABIT);
assertSetGetValidThroughputMiB(7, 7 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(7, 7 * BYTES_PER_MEGABIT);
}
@Test
@ -65,19 +70,32 @@ public class SetGetStreamThroughputTest extends CQLTester
// As part of CASSANDRA-15234 we had to do some tweaks with precision. This test has to ensure no regressions
// happen, hopefully. Internally data rate parameters values and rate limitter are set in double. Users can set
// and get only integers
assertSetGetValidThroughput(1, 0.119209289550781 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(1, 1 * BYTES_PER_MEGABIT);
assertSetGetValidThroughputMiB(1, 1 * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(1, 1 * BYTES_PER_MEGABIT);
}
@Test
public void testMaxValue()
{
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE, INTEGER_MAX_VALUE_MEGABITS_IN_MEBIBYTES * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetGetValidThroughput(MAX_INT_CONFIG_VALUE_IN_MBIT, INTEGER_MAX_VALUE_MEGABITS_IN_BYTES);
assertSetGetValidThroughputMiB(MAX_INT_CONFIG_VALUE_MIB, MAX_INT_CONFIG_VALUE_MIB * StreamRateLimiter.BYTES_PER_MEBIBYTE);
assertSetMbitGetMibValidThroughput(MAX_INT_CONFIG_VALUE_IN_MBIT, INTEGER_MAX_VALUE_MEGABITS_IN_BYTES);
}
@Test
public void testUpperBound()
{
assertSetInvalidThroughputMib(String.valueOf(Integer.MAX_VALUE));
assertSetInvalidThroughputMbit(String.valueOf(Integer.MAX_VALUE));
}
@Test
public void testZero()
{
assertSetGetValidThroughput(0, Double.MAX_VALUE);
assertSetGetValidThroughputMiB(0, Double.MAX_VALUE);
assertSetMbitGetMibValidThroughput(0, Double.MAX_VALUE);
}
@Test
@ -85,6 +103,9 @@ public class SetGetStreamThroughputTest extends CQLTester
{
assertSetInvalidThroughput("1.2", "stream_throughput: can not convert \"1.2\" to a int");
assertSetInvalidThroughput("value", "stream_throughput: can not convert \"value\" to a int");
assertSetGetMoreFlagsIsInvalid();
assertDFlagNeeded();
assertSetGetMoreFlagsIsInvalid();
}
private static void assertSetGetValidThroughput(int throughput, double rateInBytes)
@ -98,6 +119,39 @@ public class SetGetStreamThroughputTest extends CQLTester
assertThat(StreamRateLimiter.getRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.04));
}
private static void assertSetGetValidThroughputMiB(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-m", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughputMiB(throughput);
assertThat(StreamRateLimiter.getRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertDFlagNeeded()
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-m", String.valueOf(1));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
tool = invokeNodetool("getstreamthroughput");
assertThat(tool.getExitCode()).isEqualTo(2);
assertThat(tool.getStderr()).contains("Use the -d flag to quiet this error and get the exact throughput in megabits/s");
}
private static void assertSetMbitGetMibValidThroughput(int throughput, double rateInBytes)
{
ToolResult tool = invokeNodetool("setstreamthroughput", String.valueOf(throughput));
tool.assertOnCleanExit();
assertThat(tool.getStdout()).isEmpty();
assertGetThroughputMiB(throughput * MEBIBYTES_PER_MEGABIT);
assertThat(StreamRateLimiter.getRateLimiterRateInBytes()).isEqualTo(rateInBytes, withPrecision(0.01));
}
private static void assertSetInvalidThroughput(String throughput, String expectedErrorMessage)
{
ToolResult tool = throughput == null ? invokeNodetool("setstreamthroughput")
@ -106,13 +160,63 @@ public class SetGetStreamThroughputTest extends CQLTester
assertThat(tool.getStdout()).contains(expectedErrorMessage);
}
private static void assertSetInvalidThroughputMib(String throughput)
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-m", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("stream_throughput_outbound: 2147483647 is too large; it should be less " +
"than 2147483647 in megabits/s");
}
private static void assertSetInvalidThroughputMbit(String throughput)
{
ToolResult tool = invokeNodetool("setstreamthroughput", throughput);
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("Invalid data rate: 2147483647 megabits per second; stream_throughput_outbound " +
"and inter_dc_stream_throughput_outbound should be between 0 and 2147483646 in megabits per second");
}
private static void assertSetGetMoreFlagsIsInvalid()
{
ToolResult tool = invokeNodetool("setstreamthroughput", "-m", "5", "-e", "5");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use -e and -m at the same time");
tool = invokeNodetool("getstreamthroughput", "-m", "-e");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getstreamthroughput", "-m", "-d");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getstreamthroughput", "-d", "-e");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
tool = invokeNodetool("getstreamthroughput", "-m", "-e", "-d");
assertThat(tool.getExitCode()).isEqualTo(1);
assertThat(tool.getStdout()).contains("You cannot use more than one flag with this command");
}
private static void assertGetThroughput(int expected)
{
ToolResult tool = invokeNodetool("getstreamthroughput");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current stream throughput: " + expected + " megabits per second");
assertThat(tool.getStdout()).contains("Current stream throughput: " + expected + " Mb/s");
else
assertThat(tool.getStdout()).contains("Current stream throughput: unlimited");
}
private static void assertGetThroughputMiB(double expected)
{
ToolResult tool = invokeNodetool("getstreamthroughput", "-m");
tool.assertOnCleanExit();
if (expected > 0)
assertThat(tool.getStdout()).contains("Current stream throughput: " + expected + " MiB/s");
else
assertThat(tool.getStdout()).contains("Current stream throughput: unlimited");
}

25
tools/stress/src/org/apache/cassandra/io/sstable/StressCQLSSTableWriter.java
View File

@ -501,13 +501,13 @@ public class StressCQLSSTableWriter implements Closeable
* The size of the buffer to use.
* <p>
* This defines how much data will be buffered before being written as
* a new SSTable. This correspond roughly to the data size that will have the created
* a new SSTable. This corresponds roughly to the data size that will have the created
* sstable.
* <p>
* The default is 128MiB, which should be reasonable for a 1GB heap. If you experience
* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience
* OOM while using the writer, you should lower this value.
*
* @param size the size to use in MB.
* @param size the size to use in MiB.
* @return this builder.
*/
public Builder withBufferSizeInMiB(int size)
@ -516,6 +516,25 @@ public class StressCQLSSTableWriter implements Closeable
return this;
}
/**
* This method is deprecated in favor of the new withBufferSizeInMiB(int size)
* The size of the buffer to use.
* <p>
* This defines how much data will be buffered before being written as
* a new SSTable. This corresponds roughly to the data size that will have the created
* sstable.
* <p>
* The default is 128MiB, which should be reasonable for a 1GiB heap. If you experience
* OOM while using the writer, you should lower this value.
*
* @param size the size to use in MiB.
* @return this builder.
*/
public Builder withBufferSizeInMB(int size)
{
return withBufferSizeInMiB(size);
}
/**
* Creates a StressCQLSSTableWriter that expects sorted inputs.
* <p>

3
tools/stress/src/org/apache/cassandra/stress/CompactionStress.java
View File

@ -36,7 +36,6 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
@ -218,7 +217,7 @@ public abstract class CompactionStress implements Runnable
//Setup
CompactionManager.instance.setMaximumCompactorThreads(threads);
CompactionManager.instance.setCoreCompactorThreads(threads);
CompactionManager.instance.setRate(0);
CompactionManager.instance.setRateInBytes(0);
StressProfile stressProfile = getStressProfile();
ColumnFamilyStore cfs = initCf(stressProfile, true);