This is an automated email from the ASF dual-hosted git repository. bdeggleston pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 70e6f2952fd648f45aa9075909c13e5783141318 Merge: cbf4da4 1268530 Author: Blake Eggleston <bdeggles...@gmail.com> AuthorDate: Thu Feb 14 13:21:54 2019 -0800 Merge branch 'cassandra-3.11' into trunk CHANGES.txt | 1 + NEWS.txt | 3 + conf/cassandra.yaml | 11 + src/java/org/apache/cassandra/config/Config.java | 5 + .../cassandra/config/DatabaseDescriptor.java | 54 ++++ .../org/apache/cassandra/repair/RepairJob.java | 4 +- .../org/apache/cassandra/repair/RepairSession.java | 17 +- .../apache/cassandra/repair/ValidationManager.java | 17 +- .../cassandra/service/ActiveRepairService.java | 12 + .../service/ActiveRepairServiceMBean.java | 3 + .../apache/cassandra/service/StorageService.java | 11 +- .../cassandra/service/StorageServiceMBean.java | 4 + .../org/apache/cassandra/utils/MerkleTree.java | 46 +++ .../cassandra/config/DatabaseDescriptorTest.java | 69 +++++ .../org/apache/cassandra/repair/RepairJobTest.java | 317 +++++++++++++++++---- .../org/apache/cassandra/repair/ValidatorTest.java | 158 +++++++++- .../org/apache/cassandra/utils/MerkleTreeTest.java | 98 ++++++- 17 files changed, 762 insertions(+), 68 deletions(-) diff --cc CHANGES.txt index 87f3e2b,00ca115..1394ea8 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -354,6 -11,6 +354,7 @@@ * Make stop-server.bat wait for Cassandra to terminate (CASSANDRA-14829) * Correct sstable sorting for garbagecollect and levelled compaction (CASSANDRA-14870) Merged from 3.0: ++ * Improve merkle tree size and time on heap (CASSANDRA-14096) * Severe concurrency issues in STCS,DTCS,TWCS,TMD.Topology,TypeParser * Add a script to make running the cqlsh tests in cassandra repo easier (CASSANDRA-14951) * If SizeEstimatesRecorder misses a 'onDropTable' notification, the size_estimates table will never be cleared for that table. (CASSANDRA-14905) diff --cc NEWS.txt index 00c3178,d5a9128..d9950e3 --- a/NEWS.txt +++ b/NEWS.txt @@@ -113,112 -47,8 +113,115 @@@ New feature Upgrading --------- - - repair_session_max_tree_depth setting has been added to cassandra.yaml to allow operators to reduce - merkle tree size if repair is creating too much heap pressure. See CASSANDRA-14096 for details. + - CASSANDRA-13241 lowered the default chunk_lengh_in_kb for compresesd tables from + 64kb to 16kb. For highly compressible data this can have a noticeable impact + on space utilization. You may want to consider manually specifying this value. + - Additional columns have been added to system_distributed.repair_history, + system_traces.sessions and system_traces.events. As a result select queries + againsts these tables will fail and generate an error in the log + during upgrade when the cluster is mixed version. The tables can be made + readable by following the instructions in CASSANDRA-14897 to add the + new columns to the system tables before upgrading. + - Timestamp ties between values resolve differently: if either value has a TTL, + this value always wins. This is to provide consistent reconciliation before + and after the value expires into a tombstone. + - Cassandra 4.0 removed support for COMPACT STORAGE tables. All Compact Tables + have to be migrated using `ALTER ... DROP COMPACT STORAGE` statement in 3.0/3.11. + Cassandra starting 4.0 will not start if flags indicate that the table is non-CQL. + Syntax for creating compact tables is also deprecated. + - Support for legacy auth tables in the system_auth keyspace (users, + permissions, credentials) and the migration code has been removed. Migration + of these legacy auth tables must have been completed before the upgrade to + 4.0 and the legacy tables must have been removed. See the 'Upgrading' section + for version 2.2 for migration instructions. + - Cassandra 4.0 removed support for the deprecated Thrift interface. Amongst + other things, this implies the removal of all yaml options related to thrift + ('start_rpc', rpc_port, ...). + - Cassandra 4.0 removed support for any pre-3.0 format. This means you + cannot upgrade from a 2.x version to 4.0 directly, you have to upgrade to + a 3.0.x/3.x version first (and run upgradesstable). In particular, this + mean Cassandra 4.0 cannot load or read pre-3.0 sstables in any way: you + will need to upgrade those sstable in 3.0.x/3.x first. + - Upgrades from 3.0.x or 3.x are supported since 3.0.13 or 3.11.0, previous + versions will causes issues during rolling upgrades (CASSANDRA-13274). + - Cassandra will no longer allow invalid keyspace replication options, such + as invalid datacenter names for NetworkTopologyStrategy. Operators MUST + add new nodes to a datacenter before they can set set ALTER or CREATE + keyspace replication policies using that datacenter. Existing keyspaces + will continue to operate, but CREATE and ALTER will validate that all + datacenters specified exist in the cluster. + - Cassandra 4.0 fixes a problem with incremental repair which caused repaired + data to be inconsistent between nodes. The fix changes the behavior of both + full and incremental repairs. For full repairs, data is no longer marked + repaired. For incremental repairs, anticompaction is run at the beginning + of the repair, instead of at the end. If incremental repair was being used + prior to upgrading, a full repair should be run after upgrading to resolve + any inconsistencies. + - Config option index_interval has been removed (it was deprecated since 2.0) + - Deprecated repair JMX APIs are removed. + - The version of snappy-java has been upgraded to 1.1.2.6 + - the miniumum value for internode message timeouts is 10ms. Previously, any + positive value was allowed. See cassandra.yaml entries like + read_request_timeout_in_ms for more details. + - Cassandra 4.0 allows a single port to be used for both secure and insecure + connections between cassandra nodes (CASSANDRA-10404). See the yaml for + specific property changes, and see the security doc for full details. + - Due to the parallelization of the initial build of materialized views, + the per token range view building status is stored in the new table + `system.view_builds_in_progress`. The old table `system.views_builds_in_progress` + is no longer used and can be removed. See CASSANDRA-12245 for more details. + - Config option commitlog_sync_batch_window_in_ms has been deprecated as it's + documentation has been incorrect and the setting itself near useless. + Batch mode remains a valid commit log mode, however. + - There is a new commit log mode, group, which is similar to batch mode + but blocks for up to a configurable number of milliseconds between disk flushes. + - nodetool clearsnapshot now required the --all flag to remove all snapshots. + Previous behavior would delete all snapshots by default. + - Nodes are now identified by a combination of IP, and storage port. + Existing JMX APIs, nodetool, and system tables continue to work + and accept/return just an IP, but there is a new + version of each that works with the full unambiguous identifier. + You should prefer these over the deprecated ambiguous versions that only + work with an IP. This was done to support multiple instances per IP. + Additionally we are moving to only using a single port for encrypted and + unencrypted traffic and if you want multiple instances per IP you must + first switch encrypted traffic to the storage port and not a separate + encrypted port. If you want to use multiple instances per IP + with SSL you will need to use StartTLS on storage_port and set + outgoing_encrypted_port_source to gossip outbound connections + know what port to connect to for each instance. Before changing + storage port or native port at nodes you must first upgrade the entire cluster + and clients to 4.0 so they can handle the port not being consistent across + the cluster. + - Names of AWS regions/availability zones have been cleaned up to more correctly + match the Amazon names. There is now a new option in conf/cassandra-rackdc.properties + that lets users enable the correct names for new clusters, or use the legacy + names for existing clusters. See conf/cassandra-rackdc.properties for details. + - Background repair has been removed. dclocal_read_repair_chance and + read_repair_chance table options have been removed and are now rejected. + See CASSANDRA-13910 for details. + - Internode TCP connections that do not ack segments for 30s will now + be automatically detected and closed via the Linux TCP_USER_TIMEOUT + socket option. This should be exceedingly rare, but AWS networks (and + other stateful firewalls) apparently suffer from this issue. You can + tune the timeouts on TCP connection and segment ack via the + `cassandra.yaml:internode_tcp_connect_timeout_in_ms` and + `cassandra.yaml:internode_tcp_user_timeout_in_ms` options respectively. + See CASSANDRA-14358 for details. ++ - repair_session_space_in_mb setting has been added to cassandra.yaml to allow operators to reduce ++ merkle tree size if repair is creating too much heap pressure. The repair_session_max_tree_depth ++ setting added in 3.0.19 and 3.11.5 is deprecated in favor of this setting. See CASSANDRA-14096 + +Materialized Views +------------------- + - Following a discussion regarding concerns about the design and safety of Materialized Views, the C* development + community no longer recommends them for production use, and considers them experimental. Warnings messages will + now be logged when they are created. (See https://www.mail-archive.com/dev@cassandra.apache.org/msg11511.html) + - An 'enable_materialized_views' flag has been added to cassandra.yaml to allow operators to prevent creation of + views + - CREATE MATERIALIZED VIEW syntax has become stricter. Partition key columns are no longer implicitly considered + to be NOT NULL, and no base primary key columns get automatically included in view definition. You have to + specify them explicitly now. 3.11.4 ====== diff --cc conf/cassandra.yaml index dde4296,a263d8a..78fb162 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@@ -506,6 -498,19 +506,17 @@@ concurrent_materialized_view_writes: 3 # off heap objects memtable_allocation_type: heap_buffers -# Limits the maximum Merkle tree depth to avoid consuming too much -# memory during repairs. -# -# The default setting of 18 generates trees of maximum size around -# 50 MiB / tree. If you are running out of memory during repairs consider -# lowering this to 15 (~6 MiB / tree) or lower, but try not to lower it -# too much past that or you will lose too much resolution and stream -# too much redundant data during repair. Cannot be set lower than 10. ++# Limit memory usage for Merkle tree calculations during repairs. The default ++# 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 ++# this or using subrange repair. + # + # For more details see https://issues.apache.org/jira/browse/CASSANDRA-14096. + # -# repair_session_max_tree_depth: 18 ++# repair_session_space_in_mb: + # Total space to use for commit logs on disk. # # If space gets above this value, Cassandra will flush every dirty CF diff --cc src/java/org/apache/cassandra/config/Config.java index 4a55b2e,528cf4f..a95db23 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@@ -123,6 -123,9 +123,11 @@@ public class Confi public Integer memtable_offheap_space_in_mb; public Float memtable_cleanup_threshold = null; + // Limit the maximum depth of repair session merkle trees - public volatile int repair_session_max_tree_depth = 18; ++ @Deprecated ++ public volatile Integer repair_session_max_tree_depth = null; ++ public volatile Integer repair_session_space_in_mb = null; + public int storage_port = 7000; public int ssl_storage_port = 7001; public String listen_address; diff --cc src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 5c70c9a,069a17e..3f80d71 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@@ -461,7 -437,13 +461,28 @@@ public class DatabaseDescripto else logger.info("Global memtable off-heap threshold is enabled at {}MB", conf.memtable_offheap_space_in_mb); - if (conf.repair_session_max_tree_depth < 10) - throw new ConfigurationException("repair_session_max_tree_depth should not be < 10, but was " + conf.repair_session_max_tree_depth); - if (conf.repair_session_max_tree_depth > 20) - logger.warn("repair_session_max_tree_depth of " + conf.repair_session_max_tree_depth + " > 20 could lead to excessive memory usage"); ++ if (conf.repair_session_max_tree_depth != null) ++ { ++ logger.warn("repair_session_max_tree_depth has been deprecated and should be removed from cassandra.yaml. Use repair_session_space_in_mb instead"); ++ if (conf.repair_session_max_tree_depth < 10) ++ throw new ConfigurationException("repair_session_max_tree_depth should not be < 10, but was " + conf.repair_session_max_tree_depth); ++ if (conf.repair_session_max_tree_depth > 20) ++ logger.warn("repair_session_max_tree_depth of " + conf.repair_session_max_tree_depth + " > 20 could lead to excessive memory usage"); ++ } ++ else ++ { ++ conf.repair_session_max_tree_depth = 20; ++ } ++ ++ if (conf.repair_session_space_in_mb == null) ++ conf.repair_session_space_in_mb = Math.max(1, (int) (Runtime.getRuntime().maxMemory() / (16 * 1048576))); ++ ++ if (conf.repair_session_space_in_mb < 1) ++ throw new ConfigurationException("repair_session_space_in_mb must be > 0, but was " + conf.repair_session_space_in_mb); ++ else if (conf.repair_session_space_in_mb > (int) (Runtime.getRuntime().maxMemory() / (4 * 1048576))) ++ logger.warn("A repair_session_space_in_mb of " + conf.repair_session_space_in_mb + " megabytes is likely to cause heap pressure"); + - if (conf.thrift_framed_transport_size_in_mb <= 0) - throw new ConfigurationException("thrift_framed_transport_size_in_mb must be positive, but was " + conf.thrift_framed_transport_size_in_mb, false); + checkForLowestAcceptedTimeouts(conf); if (conf.native_transport_max_frame_size_in_mb <= 0) throw new ConfigurationException("native_transport_max_frame_size_in_mb must be positive, but was " + conf.native_transport_max_frame_size_in_mb, false); @@@ -2382,11 -2280,27 +2403,44 @@@ return conf.memtable_allocation_type; } - public static Float getMemtableCleanupThreshold() - { - return conf.memtable_cleanup_threshold; - } - + public static int getRepairSessionMaxTreeDepth() + { + return conf.repair_session_max_tree_depth; + } + + public static void setRepairSessionMaxTreeDepth(int depth) + { + if (depth < 10) + throw new ConfigurationException("Cannot set repair_session_max_tree_depth to " + depth + + " which is < 10, doing nothing"); + else if (depth > 20) + logger.warn("repair_session_max_tree_depth of " + depth + " > 20 could lead to excessive memory usage"); + + conf.repair_session_max_tree_depth = depth; + } + ++ public static int getRepairSessionSpaceInMegabytes() ++ { ++ return conf.repair_session_space_in_mb; ++ } ++ ++ public static void setRepairSessionSpaceInMegabytes(int sizeInMegabytes) ++ { ++ if (sizeInMegabytes < 1) ++ throw new ConfigurationException("Cannot set repair_session_space_in_mb to " + sizeInMegabytes + ++ " < 1 megabyte"); ++ else if (sizeInMegabytes > (int) (Runtime.getRuntime().maxMemory() / (4 * 1048576))) ++ logger.warn("A repair_session_space_in_mb of " + conf.repair_session_space_in_mb + ++ " megabytes is likely to cause heap pressure."); ++ ++ conf.repair_session_space_in_mb = sizeInMegabytes; ++ } ++ + public static Float getMemtableCleanupThreshold() + { + return conf.memtable_cleanup_threshold; + } + public static int getIndexSummaryResizeIntervalInMinutes() { return conf.index_summary_resize_interval_in_minutes; diff --cc src/java/org/apache/cassandra/repair/RepairJob.java index ebeb1f9,6f89a86..a67aac0 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@@ -17,13 -17,10 +17,14 @@@ */ package org.apache.cassandra.repair; -import java.net.InetAddress; import java.util.*; +import java.util.function.Predicate; +import java.util.function.Function; +import java.util.stream.Collectors; + import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@@ -239,96 -171,6 +240,97 @@@ public class RepairJob extends Abstract return syncTasks; } + private ListenableFuture<List<SyncStat>> optimisedSyncing(List<TreeResponse> trees) + { + List<SyncTask> syncTasks = createOptimisedSyncingSyncTasks(desc, + trees, + FBUtilities.getLocalAddressAndPort(), + this::isTransient, + this::getDC, + session.isIncremental, + session.previewKind); + + return executeTasks(syncTasks); + } + - private ListenableFuture<List<SyncStat>> executeTasks(List<SyncTask> syncTasks) ++ @VisibleForTesting ++ ListenableFuture<List<SyncStat>> executeTasks(List<SyncTask> syncTasks) + { + for (SyncTask task : syncTasks) + { + if (!task.isLocal()) + session.trackSyncCompletion(Pair.create(desc, task.nodePair()), (CompletableRemoteSyncTask) task); + taskExecutor.submit(task); + } + + return Futures.allAsList(syncTasks); + } + + static List<SyncTask> createOptimisedSyncingSyncTasks(RepairJobDesc desc, + List<TreeResponse> trees, + InetAddressAndPort local, + Predicate<InetAddressAndPort> isTransient, + Function<InetAddressAndPort, String> getDC, + boolean isIncremental, + PreviewKind previewKind) + { + List<SyncTask> syncTasks = new ArrayList<>(); + // We need to difference all trees one against another + DifferenceHolder diffHolder = new DifferenceHolder(trees); + + logger.debug("diffs = {}", diffHolder); + PreferedNodeFilter preferSameDCFilter = (streaming, candidates) -> + candidates.stream() + .filter(node -> getDC.apply(streaming) + .equals(getDC.apply(node))) + .collect(Collectors.toSet()); + ImmutableMap<InetAddressAndPort, HostDifferences> reducedDifferences = ReduceHelper.reduce(diffHolder, preferSameDCFilter); + + for (int i = 0; i < trees.size(); i++) + { + InetAddressAndPort address = trees.get(i).endpoint; + + // we don't stream to transient replicas + if (isTransient.test(address)) + continue; + + HostDifferences streamsFor = reducedDifferences.get(address); + if (streamsFor != null) + { + Preconditions.checkArgument(streamsFor.get(address).isEmpty(), "We should not fetch ranges from ourselves"); + for (InetAddressAndPort fetchFrom : streamsFor.hosts()) + { + List<Range<Token>> toFetch = streamsFor.get(fetchFrom); + assert !toFetch.isEmpty(); + + logger.debug("{} is about to fetch {} from {}", address, toFetch, fetchFrom); + SyncTask task; + if (address.equals(local)) + { + task = new LocalSyncTask(desc, address, fetchFrom, toFetch, isIncremental ? desc.parentSessionId : null, + true, false, previewKind); + } + else + { + task = new AsymmetricRemoteSyncTask(desc, address, fetchFrom, toFetch, previewKind); + } + syncTasks.add(task); + + } + } + else + { + logger.debug("Node {} has nothing to stream", address); + } + } + return syncTasks; + } + + private String getDC(InetAddressAndPort address) + { + return DatabaseDescriptor.getEndpointSnitch().getDatacenter(address); + } + /** * Creates {@link ValidationTask} and submit them to task executor in parallel. * diff --cc src/java/org/apache/cassandra/repair/RepairSession.java index a38205e,3d25cbf..40f3dbe --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@@ -100,13 -96,12 +101,13 @@@ public class RepairSession extends Abst private final AtomicBoolean isFailed = new AtomicBoolean(false); // Each validation task waits response from replica in validating ConcurrentMap (keyed by CF name and endpoint address) - private final ConcurrentMap<Pair<RepairJobDesc, InetAddress>, ValidationTask> validating = new ConcurrentHashMap<>(); + private final ConcurrentMap<Pair<RepairJobDesc, InetAddressAndPort>, ValidationTask> validating = new ConcurrentHashMap<>(); // Remote syncing jobs wait response in syncingTasks map - private final ConcurrentMap<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> syncingTasks = new ConcurrentHashMap<>(); + private final ConcurrentMap<Pair<RepairJobDesc, SyncNodePair>, CompletableRemoteSyncTask> syncingTasks = new ConcurrentHashMap<>(); // Tasks(snapshot, validate request, differencing, ...) are run on taskExecutor - public final ListeningExecutorService taskExecutor = MoreExecutors.listeningDecorator(DebuggableThreadPoolExecutor.createCachedThreadpoolWithMaxSize("RepairJobTask")); + public final ListeningExecutorService taskExecutor; + public final boolean optimiseStreams; private volatile boolean terminated = false; @@@ -140,38 -135,17 +141,44 @@@ this.parallelismDegree = parallelismDegree; this.keyspace = keyspace; this.cfnames = cfnames; - this.ranges = ranges; - this.endpoints = endpoints; - this.repairedAt = repairedAt; + + //If force then filter out dead endpoints + boolean forceSkippedReplicas = false; + if (force) + { + logger.debug("force flag set, removing dead endpoints"); + final Set<InetAddressAndPort> removeCandidates = new HashSet<>(); + for (final InetAddressAndPort endpoint : commonRange.endpoints) + { + if (!FailureDetector.instance.isAlive(endpoint)) + { + logger.info("Removing a dead node from Repair due to -force {}", endpoint); + removeCandidates.add(endpoint); + } + } + if (!removeCandidates.isEmpty()) + { + // we shouldn't be recording a successful repair if + // any replicas are excluded from the repair + forceSkippedReplicas = true; + Set<InetAddressAndPort> filteredEndpoints = new HashSet<>(commonRange.endpoints); + filteredEndpoints.removeAll(removeCandidates); + commonRange = new CommonRange(filteredEndpoints, commonRange.transEndpoints, commonRange.ranges); + } + } + + this.commonRange = commonRange; + this.isIncremental = isIncremental; + this.previewKind = previewKind; this.pullRepair = pullRepair; + this.skippedReplicas = forceSkippedReplicas; + this.optimiseStreams = optimiseStreams; + this.taskExecutor = MoreExecutors.listeningDecorator(createExecutor()); + } + - @VisibleForTesting + protected DebuggableThreadPoolExecutor createExecutor() + { + return DebuggableThreadPoolExecutor.createCachedThreadpoolWithMaxSize("RepairJobTask"); } public UUID getId() @@@ -228,20 -197,25 +235,26 @@@ * @param nodes nodes that completed sync * @param success true if sync succeeded */ - public void syncComplete(RepairJobDesc desc, NodePair nodes, boolean success) + public void syncComplete(RepairJobDesc desc, SyncNodePair nodes, boolean success, List<SessionSummary> summaries) { - CompletableRemoteSyncTask task = syncingTasks.get(Pair.create(desc, nodes)); - RemoteSyncTask task = syncingTasks.get(Pair.create(desc, nodes)); ++ CompletableRemoteSyncTask task = syncingTasks.remove(Pair.create(desc, nodes)); if (task == null) { assert terminated; return; } - logger.debug("[repair #{}] Repair completed between {} and {} on {}", getId(), nodes.endpoint1, nodes.endpoint2, desc.columnFamily); - task.syncComplete(success); + if (logger.isDebugEnabled()) + logger.debug("{} Repair completed between {} and {} on {}", previewKind.logPrefix(getId()), nodes.coordinator, nodes.peer, desc.columnFamily); + task.syncComplete(success, summaries); } + @VisibleForTesting - Map<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> getSyncingTasks() ++ Map<Pair<RepairJobDesc, SyncNodePair>, CompletableRemoteSyncTask> getSyncingTasks() + { + return Collections.unmodifiableMap(syncingTasks); + } + private String repairedNodes() { StringBuilder sb = new StringBuilder(); diff --cc src/java/org/apache/cassandra/repair/ValidationManager.java index d664c8a,0000000..bbd5219 mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/repair/ValidationManager.java +++ b/src/java/org/apache/cassandra/repair/ValidationManager.java @@@ -1,163 -1,0 +1,176 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.repair; + +import java.io.IOException; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + ++import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.metrics.TableMetrics; +import org.apache.cassandra.utils.FBUtilities; ++import org.apache.cassandra.utils.MerkleTree; +import org.apache.cassandra.utils.MerkleTrees; + +public class ValidationManager +{ + private static final Logger logger = LoggerFactory.getLogger(ValidationManager.class); + + public static final ValidationManager instance = new ValidationManager(); + + private ValidationManager() {} + + private static MerkleTrees createMerkleTrees(ValidationPartitionIterator validationIterator, Collection<Range<Token>> ranges, ColumnFamilyStore cfs) + { + MerkleTrees tree = new MerkleTrees(cfs.getPartitioner()); + long allPartitions = validationIterator.estimatedPartitions(); + Map<Range<Token>, Long> rangePartitionCounts = validationIterator.getRangePartitionCounts(); + ++ // The repair coordinator must hold RF trees in memory at once, so a given validation compaction can only ++ // use 1 / RF of the allowed space. ++ long availableBytes = (DatabaseDescriptor.getRepairSessionSpaceInMegabytes() * 1048576) / ++ cfs.keyspace.getReplicationStrategy().getReplicationFactor().allReplicas; ++ + for (Range<Token> range : ranges) + { + long numPartitions = rangePartitionCounts.get(range); + double rangeOwningRatio = allPartitions > 0 ? (double)numPartitions / allPartitions : 0; + // determine max tree depth proportional to range size to avoid blowing up memory with multiple tress, - // capping at 20 to prevent large tree (CASSANDRA-11390) - int maxDepth = rangeOwningRatio > 0 ? (int) Math.floor(20 - Math.log(1 / rangeOwningRatio) / Math.log(2)) : 0; ++ // capping at a depth that does not exceed our memory budget (CASSANDRA-11390, CASSANDRA-14096) ++ int rangeAvailableBytes = Math.max(1, (int) (rangeOwningRatio * availableBytes)); ++ // Try to estimate max tree depth that fits the space budget assuming hashes of 256 bits = 32 bytes ++ // note that estimatedMaxDepthForBytes cannot return a number lower than 1 ++ int estimatedMaxDepth = MerkleTree.estimatedMaxDepthForBytes(cfs.getPartitioner(), rangeAvailableBytes, 32); ++ int maxDepth = rangeOwningRatio > 0 ++ ? Math.min(estimatedMaxDepth, DatabaseDescriptor.getRepairSessionMaxTreeDepth()) ++ : 0; + // determine tree depth from number of partitions, capping at max tree depth (CASSANDRA-5263) + int depth = numPartitions > 0 ? (int) Math.min(Math.ceil(Math.log(numPartitions) / Math.log(2)), maxDepth) : 0; + tree.addMerkleTree((int) Math.pow(2, depth), range); + } + if (logger.isDebugEnabled()) + { + // MT serialize may take time + logger.debug("Created {} merkle trees with merkle trees size {}, {} partitions, {} bytes", tree.ranges().size(), tree.size(), allPartitions, MerkleTrees.serializer.serializedSize(tree, 0)); + } + + return tree; + } + + private static ValidationPartitionIterator getValidationIterator(TableRepairManager repairManager, Validator validator) throws IOException + { + RepairJobDesc desc = validator.desc; + return repairManager.getValidationIterator(desc.ranges, desc.parentSessionId, desc.sessionId, validator.isIncremental, validator.nowInSec); + } + + /** + * Performs a readonly "compaction" of all sstables in order to validate complete rows, + * but without writing the merge result + */ + @SuppressWarnings("resource") + private void doValidation(ColumnFamilyStore cfs, Validator validator) throws IOException + { + // this isn't meant to be race-proof, because it's not -- it won't cause bugs for a CFS to be dropped + // mid-validation, or to attempt to validate a droped CFS. this is just a best effort to avoid useless work, + // particularly in the scenario where a validation is submitted before the drop, and there are compactions + // started prior to the drop keeping some sstables alive. Since validationCompaction can run + // concurrently with other compactions, it would otherwise go ahead and scan those again. + if (!cfs.isValid()) + return; + + // Create Merkle trees suitable to hold estimated partitions for the given ranges. + // We blindly assume that a partition is evenly distributed on all sstables for now. + long start = System.nanoTime(); + long partitionCount = 0; + long estimatedTotalBytes = 0; + try (ValidationPartitionIterator vi = getValidationIterator(cfs.getRepairManager(), validator)) + { + MerkleTrees tree = createMerkleTrees(vi, validator.desc.ranges, cfs); + try + { + // validate the CF as we iterate over it + validator.prepare(cfs, tree); + while (vi.hasNext()) + { + try (UnfilteredRowIterator partition = vi.next()) + { + validator.add(partition); + partitionCount++; + } + } + validator.complete(); + } + finally + { + estimatedTotalBytes = vi.getEstimatedBytes(); + partitionCount = vi.estimatedPartitions(); + } + } + finally + { + cfs.metric.bytesValidated.update(estimatedTotalBytes); + cfs.metric.partitionsValidated.update(partitionCount); + } + if (logger.isDebugEnabled()) + { + long duration = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + logger.debug("Validation of {} partitions (~{}) finished in {} msec, for {}", + partitionCount, + FBUtilities.prettyPrintMemory(estimatedTotalBytes), + duration, + validator.desc); + } + } + + /** + * Does not mutate data, so is not scheduled. + */ + public Future<?> submitValidation(ColumnFamilyStore cfs, Validator validator) + { + Callable<Object> validation = new Callable<Object>() + { + public Object call() throws IOException + { + try (TableMetrics.TableTimer.Context c = cfs.metric.validationTime.time()) + { + doValidation(cfs, validator); + } + catch (Throwable e) + { + // we need to inform the remote end of our failure, otherwise it will hang on repair forever + validator.fail(); + throw e; + } + return this; + } + }; + + return cfs.getRepairManager().submitValidation(validation); + } +} diff --cc src/java/org/apache/cassandra/service/ActiveRepairService.java index 1a54e75,626aa91..525beba --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@@ -156,40 -121,8 +156,52 @@@ public class ActiveRepairService implem { this.failureDetector = failureDetector; this.gossiper = gossiper; + this.repairStatusByCmd = CacheBuilder.newBuilder() + .expireAfterWrite( + Long.getLong("cassandra.parent_repair_status_expiry_seconds", + TimeUnit.SECONDS.convert(1, TimeUnit.DAYS)), TimeUnit.SECONDS) + // using weight wouldn't work so well, since it doesn't reflect mutation of cached data + // see https://github.com/google/guava/wiki/CachesExplained + // We assume each entry is unlikely to be much more than 100 bytes, so bounding the size should be sufficient. + .maximumSize(Long.getLong("cassandra.parent_repair_status_cache_size", 100_000)) + .build(); + + MBeanWrapper.instance.registerMBean(this, MBEAN_NAME); + } + + public void start() + { + consistent.local.start(); + ScheduledExecutors.optionalTasks.scheduleAtFixedRate(consistent.local::cleanup, 0, + LocalSessions.CLEANUP_INTERVAL, + TimeUnit.SECONDS); + } + + @Override + public List<Map<String, String>> getSessions(boolean all) + { + return consistent.local.sessionInfo(all); + } + + @Override + public void failSession(String session, boolean force) + { + UUID sessionID = UUID.fromString(session); + consistent.local.cancelSession(sessionID, force); + } + ++ @Override ++ public void setRepairSessionSpaceInMegabytes(int sizeInMegabytes) ++ { ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(sizeInMegabytes); ++ } ++ ++ @Override ++ public int getRepairSessionSpaceInMegabytes() ++ { ++ return DatabaseDescriptor.getRepairSessionSpaceInMegabytes(); + } + /** * Requests repairs for the given keyspace and column families. * diff --cc src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java index 53b0acb,0000000..283d466 mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java @@@ -1,30 -1,0 +1,33 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service; + +import java.util.List; +import java.util.Map; + +public interface ActiveRepairServiceMBean +{ + public static final String MBEAN_NAME = "org.apache.cassandra.db:type=RepairService"; + + public List<Map<String, String>> getSessions(boolean all); + public void failSession(String session, boolean force); ++ ++ public void setRepairSessionSpaceInMegabytes(int sizeInMegabytes); ++ public int getRepairSessionSpaceInMegabytes(); +} diff --cc src/java/org/apache/cassandra/service/StorageService.java index ca453d0,8f4b1e7..6a57493 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@@ -3868,15 -3688,16 +3868,24 @@@ public class StorageService extends Not ActiveRepairService.instance.terminateSessions(); } - + @Nullable + public List<String> getParentRepairStatus(int cmd) + { + Pair<ActiveRepairService.ParentRepairStatus, List<String>> pair = ActiveRepairService.instance.getRepairStatus(cmd); + return pair == null ? null : + ImmutableList.<String>builder().add(pair.left.name()).addAll(pair.right).build(); + } + + public void setRepairSessionMaxTreeDepth(int depth) + { + DatabaseDescriptor.setRepairSessionMaxTreeDepth(depth); + } + + public int getRepairSessionMaxTreeDepth() + { + return DatabaseDescriptor.getRepairSessionMaxTreeDepth(); + } + /* End of MBean interface methods */ /** diff --cc src/java/org/apache/cassandra/service/StorageServiceMBean.java index e82d1ba,bcf55cf..e74f002 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@@ -357,22 -330,65 +357,26 @@@ public interface StorageServiceMBean ex */ public int repairAsync(String keyspace, Map<String, String> options); - /** - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - */ - @Deprecated - public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... tableNames) throws IOException; - - /** - * Invoke repair asynchronously. - * You can track repair progress by subscribing JMX notification sent from this StorageServiceMBean. - * Notification format is: - * type: "repair" - * userObject: int array of length 2, [0]=command number, [1]=ordinal of ActiveRepairService.Status - * - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - * - * @param parallelismDegree 0: sequential, 1: parallel, 2: DC parallel - * @return Repair command number, or 0 if nothing to repair - */ - @Deprecated - public int forceRepairAsync(String keyspace, int parallelismDegree, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... tableNames); - - /** - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - */ - @Deprecated - public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean fullRepair, String... tableNames) throws IOException; - - /** - * Same as forceRepairAsync, but handles a specified range - * - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - * - * @param parallelismDegree 0: sequential, 1: parallel, 2: DC parallel - */ - @Deprecated - public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, int parallelismDegree, Collection<String> dataCenters, Collection<String> hosts, boolean fullRepair, String... tableNames); - - /** - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - */ - @Deprecated - public int forceRepairAsync(String keyspace, boolean isSequential, boolean isLocal, boolean primaryRange, boolean fullRepair, String... tableNames); - - /** - * @deprecated use {@link #repairAsync(String keyspace, Map options)} instead. - */ - @Deprecated - public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, boolean isLocal, boolean fullRepair, String... tableNames); - public void forceTerminateAllRepairSessions(); + public void setRepairSessionMaxTreeDepth(int depth); + + public int getRepairSessionMaxTreeDepth(); + /** + * Get the status of a given parent repair session. + * @param cmd the int reference returned when issuing the repair + * @return status of parent repair from enum {@link org.apache.cassandra.repair.RepairRunnable.Status} + * followed by final message or messages of the session + */ + @Nullable + public List<String> getParentRepairStatus(int cmd); + + /** * transfer this node's data to other machines and remove it from service. + * @param force Decommission even if this will reduce N to be less than RF. */ - public void decommission() throws InterruptedException; + public void decommission(boolean force) throws InterruptedException; /** * @param newToken token to move this node to. diff --cc src/java/org/apache/cassandra/utils/MerkleTree.java index 143d839,9572a27..1d51f03 --- a/src/java/org/apache/cassandra/utils/MerkleTree.java +++ b/src/java/org/apache/cassandra/utils/MerkleTree.java @@@ -1155,4 -1190,4 +1155,50 @@@ public class MerkleTree implements Seri public TooDeep(){ super(); } } } ++ ++ /** ++ * Estimate the allowable depth while keeping the resulting heap usage of this tree under the provided ++ * number of bytes. This is important for ensuring that we do not allocate overly large trees that could ++ * OOM the JVM and cause instability. ++ * ++ * Calculated using the following logic: ++ * ++ * Let T = size of a tree of depth n ++ * ++ * T = #leafs * sizeof(leaf) + #inner * sizeof(inner) ++ * T = 2^n * L + 2^n - 1 * I ++ * ++ * T = 2^n * L + 2^n * I - I; ++ * ++ * So to solve for n given sizeof(tree_n) T: ++ * ++ * n = floor(log_2((T + I) / (L + I)) ++ * ++ * @param numBytes: The number of bytes to fit the tree within ++ * @param bytesPerHash: The number of bytes stored in a leaf node, for example 2 * murmur128 will be 256 bits ++ * or 32 bytes ++ * @return the estimated depth that will fit within the provided number of bytes ++ */ ++ public static int estimatedMaxDepthForBytes(IPartitioner partitioner, long numBytes, int bytesPerHash) ++ { ++ byte[] hashLeft = new byte[bytesPerHash]; ++ byte[] hashRigth = new byte[bytesPerHash]; ++ Leaf left = new Leaf(hashLeft); ++ Leaf right = new Leaf(hashRigth); ++ Inner inner = new Inner(partitioner.getMinimumToken(), left, right); ++ inner.calc(); ++ ++ // Some partioners have variable token sizes, try to estimate as close as we can by using the same ++ // heap estimate as the memtables use. ++ long innerTokenSize = ObjectSizes.measureDeep(partitioner.getMinimumToken()); ++ long realInnerTokenSize = partitioner.getMinimumToken().getHeapSize(); ++ ++ long sizeOfLeaf = ObjectSizes.measureDeep(left); ++ long sizeOfInner = ObjectSizes.measureDeep(inner) - ++ (ObjectSizes.measureDeep(left) + ObjectSizes.measureDeep(right) + innerTokenSize) + ++ realInnerTokenSize; ++ ++ long adjustedBytes = Math.max(1, (numBytes + sizeOfInner) / (sizeOfLeaf + sizeOfInner)); ++ return Math.max(1, (int) Math.floor(Math.log(adjustedBytes) / Math.log(2))); ++ } } diff --cc test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java index 3d88164,4788289..209c35d --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java @@@ -29,13 -31,23 +30,14 @@@ import org.junit.Assert import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; + import org.apache.cassandra.OrderedJUnit4ClassRunner; -import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.exceptions.ConfigurationException; -import org.apache.cassandra.exceptions.InvalidRequestException; -import org.apache.cassandra.gms.Gossiper; -import org.apache.cassandra.schema.KeyspaceMetadata; -import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.service.MigrationManager; -import org.apache.cassandra.thrift.ThriftConversion; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.fail; - import static org.junit.Assert.assertTrue; ++import static org.junit.Assert.fail; @RunWith(OrderedJUnit4ClassRunner.class) public class DatabaseDescriptorTest @@@ -208,46 -284,41 +210,113 @@@ assertTrue(DatabaseDescriptor.tokensFromString(null).isEmpty()); Collection<String> tokens = DatabaseDescriptor.tokensFromString(" a,b ,c , d, f,g,h"); assertEquals(7, tokens.size()); - assertTrue(tokens.containsAll(Arrays.asList(new String[]{ "a", "b", "c", "d", "f", "g", "h" }))); + assertTrue(tokens.containsAll(Arrays.asList(new String[]{"a", "b", "c", "d", "f", "g", "h"}))); + } + + @Test + public void testLowestAcceptableTimeouts() throws ConfigurationException + { + Config testConfig = new Config(); + testConfig.read_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.range_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.write_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.truncate_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.cas_contention_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.counter_write_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + testConfig.request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT + 1; + + assertTrue(testConfig.read_request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.range_request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.write_request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.truncate_request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.cas_contention_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.counter_write_request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.request_timeout_in_ms > DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + + //set less than Lowest acceptable value + testConfig.read_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.range_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.write_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.truncate_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.cas_contention_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.counter_write_request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + testConfig.request_timeout_in_ms = DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT - 1; + + DatabaseDescriptor.checkForLowestAcceptedTimeouts(testConfig); + + assertTrue(testConfig.read_request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.range_request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.write_request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.truncate_request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.cas_contention_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.counter_write_request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + assertTrue(testConfig.request_timeout_in_ms == DatabaseDescriptor.LOWEST_ACCEPTED_TIMEOUT); + } ++ ++ @Test ++ public void testRepairSessionMemorySizeToggles() ++ { ++ int previousSize = DatabaseDescriptor.getRepairSessionSpaceInMegabytes(); ++ try ++ { ++ Assert.assertEquals((Runtime.getRuntime().maxMemory() / (1024 * 1024) / 16), ++ DatabaseDescriptor.getRepairSessionSpaceInMegabytes()); ++ ++ int targetSize = (int) (Runtime.getRuntime().maxMemory() / (1024 * 1024) / 4) + 1; ++ ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(targetSize); ++ Assert.assertEquals(targetSize, DatabaseDescriptor.getRepairSessionSpaceInMegabytes()); ++ ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(10); ++ Assert.assertEquals(10, DatabaseDescriptor.getRepairSessionSpaceInMegabytes()); ++ ++ try ++ { ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(0); ++ fail("Should have received a ConfigurationException for depth of 9"); ++ } ++ catch (ConfigurationException ignored) { } ++ ++ Assert.assertEquals(10, DatabaseDescriptor.getRepairSessionSpaceInMegabytes()); ++ } ++ finally ++ { ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(previousSize); ++ } + } + + @Test + public void testRepairSessionSizeToggles() + { + int previousDepth = DatabaseDescriptor.getRepairSessionMaxTreeDepth(); + try + { - Assert.assertEquals(18, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); ++ Assert.assertEquals(20, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); + DatabaseDescriptor.setRepairSessionMaxTreeDepth(10); + Assert.assertEquals(10, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); + + try + { + DatabaseDescriptor.setRepairSessionMaxTreeDepth(9); + fail("Should have received a ConfigurationException for depth of 9"); + } + catch (ConfigurationException ignored) { } + Assert.assertEquals(10, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); + + try + { + DatabaseDescriptor.setRepairSessionMaxTreeDepth(-20); + fail("Should have received a ConfigurationException for depth of -20"); + } + catch (ConfigurationException ignored) { } + Assert.assertEquals(10, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); + + DatabaseDescriptor.setRepairSessionMaxTreeDepth(22); + Assert.assertEquals(22, DatabaseDescriptor.getRepairSessionMaxTreeDepth()); + } + finally + { + DatabaseDescriptor.setRepairSessionMaxTreeDepth(previousDepth); + } + } } diff --cc test/unit/org/apache/cassandra/repair/RepairJobTest.java index 263ecbc,e1dd5b3..6db29dc --- a/test/unit/org/apache/cassandra/repair/RepairJobTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairJobTest.java @@@ -18,608 -18,309 +18,823 @@@ package org.apache.cassandra.repair; -import java.net.InetAddress; import java.net.UnknownHostException; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; + import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; ++import java.util.concurrent.ExecutionException; + import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; + import java.util.stream.Collectors; -import com.google.common.util.concurrent.AsyncFunction; -import com.google.common.util.concurrent.Futures; +import com.google.common.collect.Sets; + import com.google.common.util.concurrent.ListenableFuture; import org.junit.After; +import org.junit.Assert; + import org.junit.Before; + import org.junit.BeforeClass; import org.junit.Test; - import org.apache.cassandra.config.DatabaseDescriptor; + import org.apache.cassandra.SchemaLoader; + import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; -import org.apache.cassandra.config.DatabaseDescriptor; + import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.dht.IPartitioner; + import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.InetAddressAndPort; + import org.apache.cassandra.net.IMessageSink; + import org.apache.cassandra.net.MessageIn; + import org.apache.cassandra.net.MessageOut; + import org.apache.cassandra.net.MessagingService; + import org.apache.cassandra.repair.messages.RepairMessage; + import org.apache.cassandra.repair.messages.SyncRequest; + import org.apache.cassandra.schema.KeyspaceParams; + import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTree; import org.apache.cassandra.utils.MerkleTrees; + import org.apache.cassandra.utils.ObjectSizes; + import org.apache.cassandra.utils.UUIDGen; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -public class RepairJobTest extends SchemaLoader +public class RepairJobTest { + private static final long TEST_TIMEOUT_S = 10; + private static final long THREAD_TIMEOUT_MILLIS = 100; + private static final IPartitioner PARTITIONER = ByteOrderedPartitioner.instance; + private static final IPartitioner MURMUR3_PARTITIONER = Murmur3Partitioner.instance; + private static final String KEYSPACE = "RepairJobTest"; + private static final String CF = "Standard1"; + private static final Object messageLock = new Object(); + ++ private static final Range<Token> range1 = range(0, 1); ++ private static final Range<Token> range2 = range(2, 3); ++ private static final Range<Token> range3 = range(4, 5); ++ private static final RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), KEYSPACE, CF, Arrays.asList()); + private static final List<Range<Token>> fullRange = Collections.singletonList(new Range<>(MURMUR3_PARTITIONER.getMinimumToken(), - MURMUR3_PARTITIONER.getRandomToken())); - private static InetAddress addr1; - private static InetAddress addr2; - private static InetAddress addr3; - private static InetAddress addr4; ++ MURMUR3_PARTITIONER.getMaximumToken())); ++ private static InetAddressAndPort addr1; ++ private static InetAddressAndPort addr2; ++ private static InetAddressAndPort addr3; ++ private static InetAddressAndPort addr4; ++ private static InetAddressAndPort addr5; + private RepairSession session; + private RepairJob job; + private RepairJobDesc sessionJobDesc; + + // So that threads actually get recycled and we can have accurate memory accounting while testing + // memory retention from CASSANDRA-14096 + private static class MeasureableRepairSession extends RepairSession + { - public MeasureableRepairSession(UUID parentRepairSession, UUID id, Collection<Range<Token>> ranges, - String keyspace,RepairParallelism parallelismDegree, Set<InetAddress> endpoints, - long repairedAt, boolean pullRepair, String... cfnames) ++ public MeasureableRepairSession(UUID parentRepairSession, UUID id, CommonRange commonRange, String keyspace, ++ RepairParallelism parallelismDegree, boolean isIncremental, boolean pullRepair, ++ boolean force, PreviewKind previewKind, boolean optimiseStreams, String... cfnames) + { - super(parentRepairSession, id, ranges, keyspace, parallelismDegree, endpoints, repairedAt, pullRepair, cfnames); ++ super(parentRepairSession, id, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair, force, previewKind, optimiseStreams, cfnames); + } + + protected DebuggableThreadPoolExecutor createExecutor() + { + DebuggableThreadPoolExecutor executor = super.createExecutor(); + executor.setKeepAliveTime(THREAD_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); - return executor; - } ++ return executor; } + } - + @BeforeClass + public static void setupClass() throws UnknownHostException + { + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE, + KeyspaceParams.simple(1), + SchemaLoader.standardCFMD(KEYSPACE, CF)); - addr1 = InetAddress.getByName("127.0.0.1"); - addr2 = InetAddress.getByName("127.0.0.2"); - addr3 = InetAddress.getByName("127.0.0.3"); - addr4 = InetAddress.getByName("127.0.0.4"); ++ addr1 = InetAddressAndPort.getByName("127.0.0.1"); ++ addr2 = InetAddressAndPort.getByName("127.0.0.2"); ++ addr3 = InetAddressAndPort.getByName("127.0.0.3"); ++ addr4 = InetAddressAndPort.getByName("127.0.0.4"); ++ addr5 = InetAddressAndPort.getByName("127.0.0.5"); + } - static InetAddressAndPort addr1; - static InetAddressAndPort addr2; - static InetAddressAndPort addr3; - static InetAddressAndPort addr4; - static InetAddressAndPort addr5; + @Before + public void setup() + { - Set<InetAddress> neighbors = new HashSet<>(Arrays.asList(addr2, addr3)); ++ Set<InetAddressAndPort> neighbors = new HashSet<>(Arrays.asList(addr2, addr3)); - static Range<Token> range1 = range(0, 1); - static Range<Token> range2 = range(2, 3); - static Range<Token> range3 = range(4, 5); - static RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), "ks", "cf", Arrays.asList()); + UUID parentRepairSession = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, FBUtilities.getBroadcastAddress(), ++ ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, FBUtilities.getBroadcastAddressAndPort(), + Collections.singletonList(Keyspace.open(KEYSPACE).getColumnFamilyStore(CF)), fullRange, false, - ActiveRepairService.UNREPAIRED_SSTABLE, false); ++ ActiveRepairService.UNREPAIRED_SSTABLE, false, PreviewKind.NONE); + - this.session = new MeasureableRepairSession(parentRepairSession, UUIDGen.getTimeUUID(), fullRange, - KEYSPACE, RepairParallelism.SEQUENTIAL, neighbors, - ActiveRepairService.UNREPAIRED_SSTABLE, false, CF); ++ this.session = new MeasureableRepairSession(parentRepairSession, UUIDGen.getTimeUUID(), ++ new CommonRange(neighbors, Collections.emptySet(), fullRange), ++ KEYSPACE, RepairParallelism.SEQUENTIAL, ++ false, false, false, ++ PreviewKind.NONE, false, CF); + + this.job = new RepairJob(session, CF); + this.sessionJobDesc = new RepairJobDesc(session.parentRepairSession, session.getId(), - session.keyspace, CF, session.getRanges()); ++ session.keyspace, CF, session.ranges()); + - DatabaseDescriptor.setBroadcastAddress(addr1); ++ FBUtilities.setBroadcastInetAddress(addr1.address); + } @After public void reset() { + ActiveRepairService.instance.terminateSessions(); + MessagingService.instance().clearMessageSinks(); + FBUtilities.reset(); - DatabaseDescriptor.setBroadcastAddress(addr1.address); } - static + /** - * Ensure we can do an end to end repair of consistent data and get the messages we expect ++ * Ensure RepairJob issues the right messages in an end to end repair of consistent data + */ + @Test - public void testEndToEndNoDifferences() throws Exception ++ public void testEndToEndNoDifferences() throws InterruptedException, ExecutionException, TimeoutException { - try - { - addr1 = InetAddressAndPort.getByName("127.0.0.1"); - addr2 = InetAddressAndPort.getByName("127.0.0.2"); - addr3 = InetAddressAndPort.getByName("127.0.0.3"); - addr4 = InetAddressAndPort.getByName("127.0.0.4"); - addr5 = InetAddressAndPort.getByName("127.0.0.5"); - DatabaseDescriptor.setBroadcastAddress(addr1.address); - } - catch (UnknownHostException e) - Map<InetAddress, MerkleTrees> mockTrees = new HashMap<>(); - mockTrees.put(FBUtilities.getBroadcastAddress(), createInitialTree(false)); ++ Map<InetAddressAndPort, MerkleTrees> mockTrees = new HashMap<>(); ++ mockTrees.put(FBUtilities.getBroadcastAddressAndPort(), createInitialTree(false)); + mockTrees.put(addr2, createInitialTree(false)); + mockTrees.put(addr3, createInitialTree(false)); + + List<MessageOut> observedMessages = new ArrayList<>(); + interceptRepairMessages(mockTrees, observedMessages); + + job.run(); + + RepairResult result = job.get(TEST_TIMEOUT_S, TimeUnit.SECONDS); + - assertEquals(3, result.stats.size()); - // Should be one RemoteSyncTask left behind (other two should be local) - assertExpectedDifferences(session.getSyncingTasks().values(), 0); ++ // Since there are no differences, there should be nothing to sync. ++ assertEquals(0, result.stats.size()); + + // RepairJob should send out SNAPSHOTS -> VALIDATIONS -> done + List<RepairMessage.Type> expectedTypes = new ArrayList<>(); + for (int i = 0; i < 3; i++) + expectedTypes.add(RepairMessage.Type.SNAPSHOT); + for (int i = 0; i < 3; i++) + expectedTypes.add(RepairMessage.Type.VALIDATION_REQUEST); + + assertEquals(expectedTypes, observedMessages.stream() + .map(k -> ((RepairMessage) k.payload).messageType) + .collect(Collectors.toList())); + } + + /** + * Regression test for CASSANDRA-14096. We should not retain memory in the RepairSession once the + * ValidationTask -> SyncTask transform is done. + */ + @Test + public void testNoTreesRetainedAfterDifference() throws Throwable + { - Map<InetAddress, MerkleTrees> mockTrees = new HashMap<>(); - mockTrees.put(FBUtilities.getBroadcastAddress(), createInitialTree(false)); - mockTrees.put(addr2, createInitialTree(true)); ++ Map<InetAddressAndPort, MerkleTrees> mockTrees = new HashMap<>(); ++ mockTrees.put(addr1, createInitialTree(true)); ++ mockTrees.put(addr2, createInitialTree(false)); + mockTrees.put(addr3, createInitialTree(false)); + - List<MessageOut> observedMessages = new ArrayList<>(); - interceptRepairMessages(mockTrees, observedMessages); - + List<TreeResponse> mockTreeResponses = mockTrees.entrySet().stream() + .map(e -> new TreeResponse(e.getKey(), e.getValue())) + .collect(Collectors.toList()); ++ List<MessageOut> messages = new ArrayList<>(); ++ interceptRepairMessages(mockTrees, messages); + - long singleTreeSize = ObjectSizes.measureDeep(mockTrees.get(addr2)); ++ long singleTreeSize = ObjectSizes.measureDeep(mockTrees.get(addr1)); + - // Use a different local address so we get all RemoteSyncs (as LocalSyncs try to reach out over the network). - List<SyncTask> syncTasks = job.createSyncTasks(mockTreeResponses, addr4); ++ // Use addr4 instead of one of the provided trees to force everything to be remote sync tasks as ++ // LocalSyncTasks try to reach over the network. ++ List<SyncTask> syncTasks = RepairJob.createStandardSyncTasks(sessionJobDesc, mockTreeResponses, ++ addr4, // local ++ noTransient(), ++ session.isIncremental, ++ session.pullRepair, ++ session.previewKind); + + // SyncTasks themselves should not contain significant memory - assertTrue(ObjectSizes.measureDeep(syncTasks) < 0.8 * singleTreeSize); ++ assertTrue(ObjectSizes.measureDeep(syncTasks) < 0.2 * singleTreeSize); + - ListenableFuture<List<SyncStat>> syncResults = Futures.transform(Futures.immediateFuture(mockTreeResponses), new AsyncFunction<List<TreeResponse>, List<SyncStat>>() - { - public ListenableFuture<List<SyncStat>> apply(List<TreeResponse> treeResponses) - { - return Futures.allAsList(syncTasks); - } - }, session.taskExecutor); ++ ListenableFuture<List<SyncStat>> syncResults = job.executeTasks(syncTasks); + - // The session can retain memory in the contained executor until the threads expire, so we wait for the threads ++ // Immediately following execution the internal execution queue should still retain the trees ++ assertTrue(ObjectSizes.measureDeep(session) > singleTreeSize); ++ ++ // The session retains memory in the contained executor until the threads expire, so we wait for the threads + // that ran the Tree -> SyncTask conversions to die and release the memory + int millisUntilFreed; + for (millisUntilFreed = 0; millisUntilFreed < TEST_TIMEOUT_S * 1000; millisUntilFreed += THREAD_TIMEOUT_MILLIS) { - e.printStackTrace(); + // The measured size of the syncingTasks, and result of the computation should be much smaller ++ TimeUnit.MILLISECONDS.sleep(THREAD_TIMEOUT_MILLIS); + if (ObjectSizes.measureDeep(session) < 0.8 * singleTreeSize) + break; - TimeUnit.MILLISECONDS.sleep(THREAD_TIMEOUT_MILLIS); } + + assertTrue(millisUntilFreed < TEST_TIMEOUT_S * 1000); + + List<SyncStat> results = syncResults.get(TEST_TIMEOUT_S, TimeUnit.SECONDS); + - assertTrue(ObjectSizes.measureDeep(results) < 0.8 * singleTreeSize); ++ assertTrue(ObjectSizes.measureDeep(results) < 0.2 * singleTreeSize); ++ ++ assertEquals(2, results.size()); ++ assertEquals(0, session.getSyncingTasks().size()); ++ assertTrue(results.stream().allMatch(s -> s.numberOfDifferences == 1)); ++ ++ assertEquals(2, messages.size()); ++ assertTrue(messages.stream().allMatch(m -> ((RepairMessage) m.payload).messageType == RepairMessage.Type.SYNC_REQUEST)); + } + + @Test + public void testCreateStandardSyncTasks() + { + testCreateStandardSyncTasks(false); + } + + @Test + public void testCreateStandardSyncTasksPullRepair() + { + testCreateStandardSyncTasks(true); + } + + public static void testCreateStandardSyncTasks(boolean pullRepair) + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "same", range2, "same", range3, "same"), + treeResponse(addr2, range1, "different", range2, "same", range3, "different"), + treeResponse(addr3, range1, "same", range2, "same", range3, "same")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr1, // local + noTransient(), // transient + false, + pullRepair, + PreviewKind.ALL)); + + Assert.assertEquals(2, tasks.size()); + + SyncTask task = tasks.get(pair(addr1, addr2)); + Assert.assertTrue(task.isLocal()); + Assert.assertTrue(((LocalSyncTask) task).requestRanges); + Assert.assertEquals(!pullRepair, ((LocalSyncTask) task).transferRanges); + Assert.assertEquals(Arrays.asList(range1, range3), task.rangesToSync); + + task = tasks.get(pair(addr2, addr3)); + Assert.assertFalse(task.isLocal()); + Assert.assertTrue(task instanceof SymmetricRemoteSyncTask); + Assert.assertEquals(Arrays.asList(range1, range3), task.rangesToSync); + + Assert.assertNull(tasks.get(pair(addr1, addr3))); + } + + @Test + public void testStandardSyncTransient() + { + // Do not stream towards transient nodes + testStandardSyncTransient(true); + testStandardSyncTransient(false); + } + + public void testStandardSyncTransient(boolean pullRepair) + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "same", range2, "same", range3, "same"), + treeResponse(addr2, range1, "different", range2, "same", range3, "different")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr1, // local + transientPredicate(addr2), + false, + pullRepair, + PreviewKind.ALL)); + + Assert.assertEquals(1, tasks.size()); + + SyncTask task = tasks.get(pair(addr1, addr2)); + Assert.assertTrue(task.isLocal()); + Assert.assertTrue(((LocalSyncTask) task).requestRanges); + Assert.assertFalse(((LocalSyncTask) task).transferRanges); + Assert.assertEquals(Arrays.asList(range1, range3), task.rangesToSync); + } + + @Test + public void testStandardSyncLocalTransient() + { + // Do not stream towards transient nodes + testStandardSyncLocalTransient(true); + testStandardSyncLocalTransient(false); + } + + public void testStandardSyncLocalTransient(boolean pullRepair) + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "same", range2, "same", range3, "same"), + treeResponse(addr2, range1, "different", range2, "same", range3, "different")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr1, // local + transientPredicate(addr1), + false, + pullRepair, + PreviewKind.ALL)); + + if (pullRepair) + { + Assert.assertTrue(tasks.isEmpty()); + return; + } + + Assert.assertEquals(1, tasks.size()); + + SyncTask task = tasks.get(pair(addr1, addr2)); + Assert.assertTrue(task.isLocal()); + Assert.assertFalse(((LocalSyncTask) task).requestRanges); + Assert.assertTrue(((LocalSyncTask) task).transferRanges); + Assert.assertEquals(Arrays.asList(range1, range3), task.rangesToSync); + } + + @Test + public void testEmptyDifference() + { + // one of the nodes is a local coordinator + testEmptyDifference(addr1, noTransient(), true); + testEmptyDifference(addr1, noTransient(), false); + testEmptyDifference(addr2, noTransient(), true); + testEmptyDifference(addr2, noTransient(), false); + testEmptyDifference(addr1, transientPredicate(addr1), true); + testEmptyDifference(addr2, transientPredicate(addr1), true); + testEmptyDifference(addr1, transientPredicate(addr1), false); + testEmptyDifference(addr2, transientPredicate(addr1), false); + testEmptyDifference(addr1, transientPredicate(addr2), true); + testEmptyDifference(addr2, transientPredicate(addr2), true); + testEmptyDifference(addr1, transientPredicate(addr2), false); + testEmptyDifference(addr2, transientPredicate(addr2), false); + + // nonlocal coordinator + testEmptyDifference(addr3, noTransient(), true); + testEmptyDifference(addr3, noTransient(), false); + testEmptyDifference(addr3, noTransient(), true); + testEmptyDifference(addr3, noTransient(), false); + testEmptyDifference(addr3, transientPredicate(addr1), true); + testEmptyDifference(addr3, transientPredicate(addr1), true); + testEmptyDifference(addr3, transientPredicate(addr1), false); + testEmptyDifference(addr3, transientPredicate(addr1), false); + testEmptyDifference(addr3, transientPredicate(addr2), true); + testEmptyDifference(addr3, transientPredicate(addr2), true); + testEmptyDifference(addr3, transientPredicate(addr2), false); + testEmptyDifference(addr3, transientPredicate(addr2), false); + } + + public void testEmptyDifference(InetAddressAndPort local, Predicate<InetAddressAndPort> isTransient, boolean pullRepair) + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "same", range2, "same", range3, "same"), + treeResponse(addr2, range1, "same", range2, "same", range3, "same")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + local, // local + isTransient, + false, + pullRepair, + PreviewKind.ALL)); + + Assert.assertTrue(tasks.isEmpty()); + } + + @Test + public void testCreateStandardSyncTasksAllDifferent() + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one", range3, "one"), + treeResponse(addr2, range1, "two", range2, "two", range3, "two"), + treeResponse(addr3, range1, "three", range2, "three", range3, "three")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr1, // local + ep -> ep.equals(addr3), // transient + false, + true, + PreviewKind.ALL)); + + Assert.assertEquals(3, tasks.size()); + SyncTask task = tasks.get(pair(addr1, addr2)); + Assert.assertTrue(task.isLocal()); + Assert.assertEquals(Arrays.asList(range1, range2, range3), task.rangesToSync); + + task = tasks.get(pair(addr2, addr3)); + Assert.assertFalse(task.isLocal()); + Assert.assertEquals(Arrays.asList(range1, range2, range3), task.rangesToSync); + + task = tasks.get(pair(addr1, addr3)); + Assert.assertTrue(task.isLocal()); + Assert.assertEquals(Arrays.asList(range1, range2, range3), task.rangesToSync); + } + + @Test + public void testCreate5NodeStandardSyncTasksWithTransient() + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one", range3, "one"), + treeResponse(addr2, range1, "two", range2, "two", range3, "two"), + treeResponse(addr3, range1, "three", range2, "three", range3, "three"), + treeResponse(addr4, range1, "four", range2, "four", range3, "four"), + treeResponse(addr5, range1, "five", range2, "five", range3, "five")); + + Predicate<InetAddressAndPort> isTransient = ep -> ep.equals(addr4) || ep.equals(addr5); + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr1, // local + isTransient, // transient + false, + true, + PreviewKind.ALL)); + + SyncNodePair[] pairs = new SyncNodePair[] {pair(addr1, addr2), + pair(addr1, addr3), + pair(addr1, addr4), + pair(addr1, addr5), + pair(addr2, addr4), + pair(addr2, addr4), + pair(addr2, addr5), + pair(addr3, addr4), + pair(addr3, addr5)}; + + for (SyncNodePair pair : pairs) + { + SyncTask task = tasks.get(pair); + // Local only if addr1 is a coordinator + assertEquals(task.isLocal(), pair.coordinator.equals(addr1)); + + boolean isRemote = !pair.coordinator.equals(addr1) && !pair.peer.equals(addr1); + boolean involvesTransient = isTransient.test(pair.coordinator) || isTransient.test(pair.peer); + assertEquals(String.format("Coordinator: %s\n, Peer: %s\n",pair.coordinator, pair.peer), + isRemote && involvesTransient, + task instanceof AsymmetricRemoteSyncTask); + + // All ranges to be synchronised + Assert.assertEquals(Arrays.asList(range1, range2, range3), task.rangesToSync); + } + } - assertEquals(3, results.size()); - // Should be two RemoteSyncTasks with ranges and one empty one - assertExpectedDifferences(new ArrayList<>(session.getSyncingTasks().values()), 1, 1, 0); + @Test + public void testLocalSyncWithTransient() + { - try - { - for (InetAddressAndPort local : new InetAddressAndPort[]{ addr1, addr2, addr3 }) - { - FBUtilities.reset(); - DatabaseDescriptor.setBroadcastAddress(local.address); - testLocalSyncWithTransient(local, false); - } - } - finally ++ for (InetAddressAndPort local : new InetAddressAndPort[]{ addr1, addr2, addr3 }) + { + FBUtilities.reset(); - DatabaseDescriptor.setBroadcastAddress(addr1.address); ++ FBUtilities.setBroadcastInetAddress(local.address); ++ testLocalSyncWithTransient(local, false); + } + } - int numDifferent = 0; - for (SyncStat stat : results) + @Test + public void testLocalSyncWithTransientPullRepair() + { - try - { - for (InetAddressAndPort local : new InetAddressAndPort[]{ addr1, addr2, addr3 }) - { - FBUtilities.reset(); - DatabaseDescriptor.setBroadcastAddress(local.address); - testLocalSyncWithTransient(local, true); - } - } - finally ++ for (InetAddressAndPort local : new InetAddressAndPort[]{ addr1, addr2, addr3 }) { - if (stat.nodes.endpoint1.equals(addr2) || stat.nodes.endpoint2.equals(addr2)) - { - assertEquals(1, stat.numberOfDifferences); - numDifferent++; - } + FBUtilities.reset(); - DatabaseDescriptor.setBroadcastAddress(addr1.address); ++ FBUtilities.setBroadcastInetAddress(local.address); ++ testLocalSyncWithTransient(local, true); + } - + } + + public static void testLocalSyncWithTransient(InetAddressAndPort local, boolean pullRepair) + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one", range3, "one"), + treeResponse(addr2, range1, "two", range2, "two", range3, "two"), + treeResponse(addr3, range1, "three", range2, "three", range3, "three"), + treeResponse(addr4, range1, "four", range2, "four", range3, "four"), + treeResponse(addr5, range1, "five", range2, "five", range3, "five")); + + Predicate<InetAddressAndPort> isTransient = ep -> ep.equals(addr4) || ep.equals(addr5); + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + local, // local + isTransient, // transient + false, + pullRepair, + PreviewKind.ALL)); + + assertEquals(9, tasks.size()); + for (InetAddressAndPort addr : new InetAddressAndPort[]{ addr1, addr2, addr3 }) + { + if (local.equals(addr)) + continue; + + LocalSyncTask task = (LocalSyncTask) tasks.get(pair(local, addr)); + assertTrue(task.requestRanges); + assertEquals(!pullRepair, task.transferRanges); + } + + LocalSyncTask task = (LocalSyncTask) tasks.get(pair(local, addr4)); + assertTrue(task.requestRanges); + assertFalse(task.transferRanges); + + task = (LocalSyncTask) tasks.get(pair(local, addr5)); + assertTrue(task.requestRanges); + assertFalse(task.transferRanges); + } + + @Test + public void testLocalAndRemoteTransient() + { + testLocalAndRemoteTransient(false); + } + + @Test + public void testLocalAndRemoteTransientPullRepair() + { + testLocalAndRemoteTransient(true); + } + + private static void testLocalAndRemoteTransient(boolean pullRepair) + { - DatabaseDescriptor.setBroadcastAddress(addr4.address); ++ FBUtilities.setBroadcastInetAddress(addr4.address); + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one", range3, "one"), + treeResponse(addr2, range1, "two", range2, "two", range3, "two"), + treeResponse(addr3, range1, "three", range2, "three", range3, "three"), + treeResponse(addr4, range1, "four", range2, "four", range3, "four"), + treeResponse(addr5, range1, "five", range2, "five", range3, "five")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createStandardSyncTasks(desc, + treeResponses, + addr4, // local + ep -> ep.equals(addr4) || ep.equals(addr5), // transient + false, + pullRepair, + PreviewKind.ALL)); + + assertNull(tasks.get(pair(addr4, addr5))); + } + + @Test + public void testOptimizedCreateStandardSyncTasksAllDifferent() + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one", range3, "one"), + treeResponse(addr2, range1, "two", range2, "two", range3, "two"), + treeResponse(addr3, range1, "three", range2, "three", range3, "three")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createOptimisedSyncingSyncTasks(desc, + treeResponses, + addr1, // local + noTransient(), + addr -> "DC1", + false, + PreviewKind.ALL)); + + for (SyncNodePair pair : new SyncNodePair[]{ pair(addr1, addr2), + pair(addr1, addr3), + pair(addr2, addr1), + pair(addr2, addr3), + pair(addr3, addr1), + pair(addr3, addr2) }) + { + assertEquals(Arrays.asList(range1, range2, range3), tasks.get(pair).rangesToSync); + } + } + + @Test + public void testOptimizedCreateStandardSyncTasks() + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "one", range2, "one"), + treeResponse(addr2, range1, "one", range2, "two"), + treeResponse(addr3, range1, "three", range2, "two")); + + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createOptimisedSyncingSyncTasks(desc, + treeResponses, + addr4, // local + noTransient(), + addr -> "DC1", + false, + PreviewKind.ALL)); + + for (SyncTask task : tasks.values()) + assertTrue(task instanceof AsymmetricRemoteSyncTask); + + assertEquals(Arrays.asList(range1), tasks.get(pair(addr1, addr3)).rangesToSync); + // addr1 can get range2 from either addr2 or addr3 but not from both + assertStreamRangeFromEither(tasks, Arrays.asList(range2), + addr1, addr2, addr3); + + assertEquals(Arrays.asList(range1), tasks.get(pair(addr2, addr3)).rangesToSync); + assertEquals(Arrays.asList(range2), tasks.get(pair(addr2, addr1)).rangesToSync); + + // addr3 can get range1 from either addr1 or addr2 but not from both + assertStreamRangeFromEither(tasks, Arrays.asList(range1), + addr3, addr2, addr1); + assertEquals(Arrays.asList(range2), tasks.get(pair(addr3, addr1)).rangesToSync); + } + + @Test + public void testOptimizedCreateStandardSyncTasksWithTransient() + { + List<TreeResponse> treeResponses = Arrays.asList(treeResponse(addr1, range1, "same", range2, "same", range3, "same"), + treeResponse(addr2, range1, "different", range2, "same", range3, "different"), + treeResponse(addr3, range1, "same", range2, "same", range3, "same")); + + RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), "ks", "cf", Arrays.asList()); + Map<SyncNodePair, SyncTask> tasks = toMap(RepairJob.createOptimisedSyncingSyncTasks(desc, + treeResponses, + addr1, // local + ep -> ep.equals(addr3), + addr -> "DC1", + false, + PreviewKind.ALL)); + + assertEquals(3, tasks.size()); + SyncTask task = tasks.get(pair(addr1, addr2)); + assertTrue(task.isLocal()); + assertElementEquals(Arrays.asList(range1, range3), task.rangesToSync); + assertTrue(((LocalSyncTask)task).requestRanges); + assertFalse(((LocalSyncTask)task).transferRanges); + + assertStreamRangeFromEither(tasks, Arrays.asList(range3), + addr2, addr1, addr3); + - task = tasks.get(pair(addr2, addr3)); - assertFalse(task.isLocal()); - assertElementEquals(Arrays.asList(range1), task.rangesToSync); ++ assertStreamRangeFromEither(tasks, Arrays.asList(range1), ++ addr2, addr1, addr3); + } + + // Asserts that ranges are streamed from one of the nodes but not from the both + public static void assertStreamRangeFromEither(Map<SyncNodePair, SyncTask> tasks, List<Range<Token>> ranges, + InetAddressAndPort target, InetAddressAndPort either, InetAddressAndPort or) + { + InetAddressAndPort streamsFrom; + InetAddressAndPort doesntStreamFrom; - if (tasks.containsKey(pair(target, either))) ++ if (tasks.containsKey(pair(target, either)) && tasks.get(pair(target, either)).rangesToSync.equals(ranges)) + { + streamsFrom = either; + doesntStreamFrom = or; } - assertEquals(2, numDifferent); + else + { + doesntStreamFrom = either; + streamsFrom = or; + } + + SyncTask task = tasks.get(pair(target, streamsFrom)); + assertTrue(task instanceof AsymmetricRemoteSyncTask); + assertElementEquals(ranges, task.rangesToSync); + assertDoesntStreamRangeFrom(tasks, ranges, target, doesntStreamFrom); + } + + public static void assertDoesntStreamRangeFrom(Map<SyncNodePair, SyncTask> tasks, List<Range<Token>> ranges, + InetAddressAndPort target, InetAddressAndPort source) + { + Set<Range<Token>> rangeSet = new HashSet<>(ranges); + SyncTask task = tasks.get(pair(target, source)); + if (task == null) + return; // Doesn't stream anything + + for (Range<Token> range : task.rangesToSync) + { + assertFalse(String.format("%s shouldn't stream %s from %s", + target, range, source), + rangeSet.contains(range)); + } + } + + public static <T> void assertElementEquals(Collection<T> col1, Collection<T> col2) + { + Set<T> set1 = new HashSet<>(col1); + Set<T> set2 = new HashSet<>(col2); + Set<T> difference = Sets.difference(set1, set2); + assertTrue("Expected empty difference but got: " + difference.toString(), + difference.isEmpty()); + } + + public static Token tk(int i) + { + return PARTITIONER.getToken(ByteBufferUtil.bytes(i)); + } + + public static Range<Token> range(int from, int to) + { + return new Range<>(tk(from), tk(to)); + } + + public static TreeResponse treeResponse(InetAddressAndPort addr, Object... rangesAndHashes) + { + MerkleTrees trees = new MerkleTrees(PARTITIONER); + for (int i = 0; i < rangesAndHashes.length; i += 2) + { + Range<Token> range = (Range<Token>) rangesAndHashes[i]; + String hash = (String) rangesAndHashes[i + 1]; + MerkleTree tree = trees.addMerkleTree(2, MerkleTree.RECOMMENDED_DEPTH, range); + tree.get(range.left).hash(hash.getBytes()); + } + + return new TreeResponse(addr, trees); + } + + public static SyncNodePair pair(InetAddressAndPort node1, InetAddressAndPort node2) + { + return new SyncNodePair(node1, node2); + } + + public static Map<SyncNodePair, SyncTask> toMap(List<SyncTask> tasks) + { + Map<SyncNodePair, SyncTask> map = new HashMap(); + for (SyncTask task : tasks) + { + SyncTask oldTask = map.put(task.nodePair, task); + Assert.assertNull(String.format("\nNode pair: %s\nOld task: %s\nNew task: %s\n", + task.nodePair, + oldTask, + task), + oldTask); + } + return map; + } + + public static Predicate<InetAddressAndPort> transientPredicate(InetAddressAndPort... transientNodes) + { + Set<InetAddressAndPort> set = new HashSet<>(); + for (InetAddressAndPort node : transientNodes) + set.add(node); + + return set::contains; } - private void assertExpectedDifferences(Collection<RemoteSyncTask> tasks, Integer ... differences) + public static Predicate<InetAddressAndPort> noTransient() { - List<Integer> expectedDifferences = new ArrayList<>(Arrays.asList(differences)); - List<Integer> observedDifferences = tasks.stream() - .map(t -> (int) t.getCurrentStat().numberOfDifferences) - .collect(Collectors.toList()); - assertEquals(expectedDifferences.size(), observedDifferences.size()); - assertTrue(expectedDifferences.containsAll(observedDifferences)); + return node -> false; } + + private MerkleTrees createInitialTree(boolean invalidate) + { + MerkleTrees tree = new MerkleTrees(MURMUR3_PARTITIONER); + tree.addMerkleTrees((int) Math.pow(2, 15), fullRange); + tree.init(); + for (MerkleTree.TreeRange r : tree.invalids()) + { + r.ensureHashInitialised(); + } + + if (invalidate) + { + // change a range in one of the trees + Token token = MURMUR3_PARTITIONER.midpoint(fullRange.get(0).left, fullRange.get(0).right); + tree.invalidate(token); + tree.get(token).hash("non-empty hash!".getBytes()); + } + + return tree; + } + - private void interceptRepairMessages(Map<InetAddress, MerkleTrees> mockTrees, ++ private void interceptRepairMessages(Map<InetAddressAndPort, MerkleTrees> mockTrees, + List<MessageOut> messageCapture) + { + MessagingService.instance().addMessageSink(new IMessageSink() + { - public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress to) ++ public boolean allowOutgoingMessage(MessageOut message, int id, InetAddressAndPort to) + { + if (message == null || !(message.payload instanceof RepairMessage)) + return false; + + // So different Thread's messages don't overwrite each other. + synchronized (messageLock) + { + messageCapture.add(message); + } + + RepairMessage rm = (RepairMessage) message.payload; + switch (rm.messageType) + { + case SNAPSHOT: + MessageIn<?> messageIn = MessageIn.create(to, null, + Collections.emptyMap(), + MessagingService.Verb.REQUEST_RESPONSE, + MessagingService.current_version); + MessagingService.instance().receive(messageIn, id); + break; + case VALIDATION_REQUEST: + session.validationComplete(sessionJobDesc, to, mockTrees.get(to)); + break; + case SYNC_REQUEST: + SyncRequest syncRequest = (SyncRequest) rm; - session.syncComplete(sessionJobDesc, new NodePair(syncRequest.src, syncRequest.dst), true); ++ session.syncComplete(sessionJobDesc, new SyncNodePair(syncRequest.src, syncRequest.dst), ++ true, Collections.emptyList()); + break; + default: + break; + } + return false; + } + + public boolean allowIncomingMessage(MessageIn message, int id) + { + return message.verb == MessagingService.Verb.REQUEST_RESPONSE; + } + }); + } } diff --cc test/unit/org/apache/cassandra/repair/ValidatorTest.java index 3b582a9,9c32cef..ff6b11c --- a/test/unit/org/apache/cassandra/repair/ValidatorTest.java +++ b/test/unit/org/apache/cassandra/repair/ValidatorTest.java @@@ -17,22 -17,22 +17,24 @@@ */ package org.apache.cassandra.repair; --import java.net.InetAddress; +import java.nio.ByteBuffer; ++import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Iterator; ++import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.SettableFuture; +import com.google.common.hash.Hasher; --import org.apache.cassandra.db.compaction.CompactionManager; ++import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.compaction.CompactionsTest; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.junit.After; ++import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; @@@ -70,6 -68,6 +72,7 @@@ import static org.junit.Assert.assertTr public class ValidatorTest { private static final long TEST_TIMEOUT = 60; //seconds ++ private static int testSizeMegabytes; private static final String keyspace = "ValidatorTest"; private static final String columnFamily = "Standard1"; @@@ -82,13 -80,13 +85,21 @@@ SchemaLoader.createKeyspace(keyspace, KeyspaceParams.simple(1), SchemaLoader.standardCFMD(keyspace, columnFamily)); - partitioner = Schema.instance.getCFMetaData(keyspace, columnFamily).partitioner; + partitioner = Schema.instance.getTableMetadata(keyspace, columnFamily).partitioner; ++ testSizeMegabytes = DatabaseDescriptor.getRepairSessionSpaceInMegabytes(); } @After public void tearDown() { MessagingService.instance().clearMessageSinks(); ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(testSizeMegabytes); ++ } ++ ++ @Before ++ public void setup() ++ { ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(testSizeMegabytes); } @Test @@@ -217,45 -215,6 +228,188 @@@ assertEquals(trees.rowCount(), n); } ++ /* ++ * Test for CASSANDRA-14096 size limiting. We: ++ * 1. Limit the size of a repair session ++ * 2. Submit a validation ++ * 3. Check that the resulting tree is of limited depth ++ */ ++ @Test ++ public void testSizeLimiting() throws Exception ++ { ++ Keyspace ks = Keyspace.open(keyspace); ++ ColumnFamilyStore cfs = ks.getColumnFamilyStore(columnFamily); ++ cfs.clearUnsafe(); ++ ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(1); ++ ++ // disable compaction while flushing ++ cfs.disableAutoCompaction(); ++ ++ // 2 ** 14 rows would normally use 2^14 leaves, but with only 1 meg we should only use 2^12 ++ CompactionsTest.populate(keyspace, columnFamily, 0, 1 << 14, 0); ++ ++ cfs.forceBlockingFlush(); ++ assertEquals(1, cfs.getLiveSSTables().size()); ++ ++ // wait enough to force single compaction ++ TimeUnit.SECONDS.sleep(5); ++ ++ SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); ++ UUID repairSessionId = UUIDGen.getTimeUUID(); ++ final RepairJobDesc desc = new RepairJobDesc(repairSessionId, UUIDGen.getTimeUUID(), cfs.keyspace.getName(), ++ cfs.getTableName(), Collections.singletonList(new Range<>(sstable.first.getToken(), ++ sstable.last.getToken()))); ++ ++ ActiveRepairService.instance.registerParentRepairSession(repairSessionId, FBUtilities.getBroadcastAddressAndPort(), ++ Collections.singletonList(cfs), desc.ranges, false, ActiveRepairService.UNREPAIRED_SSTABLE, ++ false, PreviewKind.NONE); ++ ++ final CompletableFuture<MessageOut> outgoingMessageSink = registerOutgoingMessageSink(); ++ Validator validator = new Validator(desc, FBUtilities.getBroadcastAddressAndPort(), 0, true, false, PreviewKind.NONE); ++ ValidationManager.instance.submitValidation(cfs, validator); ++ ++ MessageOut message = outgoingMessageSink.get(TEST_TIMEOUT, TimeUnit.SECONDS); ++ MerkleTrees trees = ((ValidationComplete) message.payload).trees; ++ ++ Iterator<Map.Entry<Range<Token>, MerkleTree>> iterator = trees.iterator(); ++ int numTrees = 0; ++ while (iterator.hasNext()) ++ { ++ assertEquals(1 << 12, iterator.next().getValue().size(), 0.0); ++ numTrees++; ++ } ++ assertEquals(1, numTrees); ++ ++ assertEquals(trees.rowCount(), 1 << 14); ++ } ++ ++ /* ++ * Test for CASSANDRA-11390. When there are multiple subranges the trees should ++ * automatically size down to make each subrange fit in the provided memory ++ * 1. Limit the size of all the trees ++ * 2. Submit a validation against more than one range ++ * 3. Check that we have the right number and sizes of trees ++ */ ++ @Test ++ public void testRangeSplittingTreeSizeLimit() throws Exception ++ { ++ Keyspace ks = Keyspace.open(keyspace); ++ ColumnFamilyStore cfs = ks.getColumnFamilyStore(columnFamily); ++ cfs.clearUnsafe(); ++ ++ DatabaseDescriptor.setRepairSessionSpaceInMegabytes(1); ++ ++ // disable compaction while flushing ++ cfs.disableAutoCompaction(); ++ ++ // 2 ** 14 rows would normally use 2^14 leaves, but with only 1 meg we should only use 2^12 ++ CompactionsTest.populate(keyspace, columnFamily, 0, 1 << 14, 0); ++ ++ cfs.forceBlockingFlush(); ++ assertEquals(1, cfs.getLiveSSTables().size()); ++ ++ // wait enough to force single compaction ++ TimeUnit.SECONDS.sleep(5); ++ ++ SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); ++ UUID repairSessionId = UUIDGen.getTimeUUID(); ++ ++ List<Range<Token>> ranges = splitHelper(new Range<>(sstable.first.getToken(), sstable.last.getToken()), 2); ++ ++ ++ final RepairJobDesc desc = new RepairJobDesc(repairSessionId, UUIDGen.getTimeUUID(), cfs.keyspace.getName(), ++ cfs.getTableName(), ranges); ++ ++ ActiveRepairService.instance.registerParentRepairSession(repairSessionId, FBUtilities.getBroadcastAddressAndPort(), ++ Collections.singletonList(cfs), desc.ranges, false, ActiveRepairService.UNREPAIRED_SSTABLE, ++ false, PreviewKind.NONE); ++ ++ final CompletableFuture<MessageOut> outgoingMessageSink = registerOutgoingMessageSink(); ++ Validator validator = new Validator(desc, FBUtilities.getBroadcastAddressAndPort(), 0, true, false, PreviewKind.NONE); ++ ValidationManager.instance.submitValidation(cfs, validator); ++ ++ MessageOut message = outgoingMessageSink.get(TEST_TIMEOUT, TimeUnit.SECONDS); ++ MerkleTrees trees = ((ValidationComplete) message.payload).trees; ++ ++ // Should have 4 trees each with a depth of on average 10 (since each range should have gotten 0.25 megabytes) ++ Iterator<Map.Entry<Range<Token>, MerkleTree>> iterator = trees.iterator(); ++ int numTrees = 0; ++ double totalResolution = 0; ++ while (iterator.hasNext()) ++ { ++ long size = iterator.next().getValue().size(); ++ // So it turns out that sstable range estimates are pretty variable, depending on the sampling we can ++ // get a wide range of values here. So we just make sure that we're smaller than in the single range ++ // case and have the right total size. ++ assertTrue(size <= (1 << 11)); ++ assertTrue(size >= (1 << 9)); ++ totalResolution += size; ++ numTrees += 1; ++ } ++ ++ assertEquals(trees.rowCount(), 1 << 14); ++ assertEquals(4, numTrees); ++ ++ // With a single tree and a megabyte we should had a total resolution of 2^12 leaves; with multiple ++ // ranges we should get similar overall resolution, but not more. ++ assertTrue(totalResolution > (1 << 11) && totalResolution < (1 << 13)); ++ } ++ ++ private List<Range<Token>> splitHelper(Range<Token> range, int depth) ++ { ++ if (depth <= 0) ++ { ++ List<Range<Token>> tokens = new ArrayList<>(); ++ tokens.add(range); ++ return tokens; ++ } ++ Token midpoint = partitioner.midpoint(range.left, range.right); ++ List<Range<Token>> left = splitHelper(new Range<>(range.left, midpoint), depth - 1); ++ List<Range<Token>> right = splitHelper(new Range<>(midpoint, range.right), depth - 1); ++ left.addAll(right); ++ return left; ++ } ++ + @Test + public void testCountingHasher() + { + Hasher [] hashers = new Hasher[] {new Validator.CountingHasher(), Validator.CountingHasher.hashFunctions[0].newHasher(), Validator.CountingHasher.hashFunctions[1].newHasher() }; + byte [] random = UUIDGen.getTimeUUIDBytes(); + + // call all overloaded methods: + for (Hasher hasher : hashers) + { + hasher.putByte((byte) 33) + .putBytes(random) + .putBytes(ByteBuffer.wrap(random)) + .putBytes(random, 0, 3) + .putChar('a') + .putBoolean(false) + .putDouble(3.3) + .putInt(77) + .putFloat(99) + .putLong(101) + .putShort((short) 23); + } + + long len = Byte.BYTES + + random.length * 2 // both the byte[] and the ByteBuffer + + 3 // 3 bytes from the random byte[] + + Character.BYTES + + Byte.BYTES + + Double.BYTES + + Integer.BYTES + + Float.BYTES + + Long.BYTES + + Short.BYTES; + + byte [] h = hashers[0].hash().asBytes(); + assertTrue(Arrays.equals(hashers[1].hash().asBytes(), Arrays.copyOfRange(h, 0, 16))); + assertTrue(Arrays.equals(hashers[2].hash().asBytes(), Arrays.copyOfRange(h, 16, 32))); + assertEquals(len, ((Validator.CountingHasher)hashers[0]).getCount()); + } + private CompletableFuture<MessageOut> registerOutgoingMessageSink() { final CompletableFuture<MessageOut> future = new CompletableFuture<>(); diff --cc test/unit/org/apache/cassandra/utils/MerkleTreeTest.java index 64aea24,64aea24..c213271 --- a/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java +++ b/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java @@@ -19,14 -19,14 +19,18 @@@ package org.apache.cassandra.utils; import java.math.BigInteger; ++import java.nio.ByteBuffer; import java.util.*; import com.google.common.collect.Lists; ++import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.apache.cassandra.config.DatabaseDescriptor; ++import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.dht.IPartitioner; ++import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.dht.RandomPartitioner.BigIntegerToken; import org.apache.cassandra.dht.Range; @@@ -62,7 -62,7 +66,7 @@@ public class MerkleTreeTes } @Before -- public void clear() ++ public void setup() { TOKEN_SCALE = new BigInteger("8"); partitioner = RandomPartitioner.instance; @@@ -92,7 -92,7 +96,7 @@@ { if (i == -1) return new BigIntegerToken(new BigInteger("-1")); -- BigInteger bint = RandomPartitioner.MAXIMUM.divide(TOKEN_SCALE).multiply(new BigInteger(""+i)); ++ BigInteger bint = RandomPartitioner.MAXIMUM.divide(TOKEN_SCALE).multiply(new BigInteger("" + i)); return new BigIntegerToken(bint); } @@@ -113,10 -113,10 +117,10 @@@ assertEquals(new Range<>(tok(6), tok(7)), mt.get(tok(7))); // check depths -- assertEquals((byte)1, mt.get(tok(4)).depth); -- assertEquals((byte)2, mt.get(tok(6)).depth); -- assertEquals((byte)3, mt.get(tok(7)).depth); -- assertEquals((byte)3, mt.get(tok(-1)).depth); ++ assertEquals((byte) 1, mt.get(tok(4)).depth); ++ assertEquals((byte) 2, mt.get(tok(6)).depth); ++ assertEquals((byte) 3, mt.get(tok(7)).depth); ++ assertEquals((byte) 3, mt.get(tok(-1)).depth); try { @@@ -132,7 -132,7 +136,7 @@@ @Test public void testSplitLimitDepth() { -- mt = new MerkleTree(partitioner, fullRange(), (byte)2, Integer.MAX_VALUE); ++ mt = new MerkleTree(partitioner, fullRange(), (byte) 2, Integer.MAX_VALUE); assertTrue(mt.split(tok(4))); assertTrue(mt.split(tok(2))); @@@ -472,7 -472,7 +476,7 @@@ List<TreeRange> diffs = MerkleTree.difference(ltree, rtree); assertEquals(Lists.newArrayList(range), diffs); -- assertEquals(MerkleTree.FULLY_INCONSISTENT, MerkleTree.differenceHelper(ltree, rtree, new ArrayList<>(), new MerkleTree.TreeDifference(ltree.fullRange.left, ltree.fullRange.right, (byte)0))); ++ assertEquals(MerkleTree.FULLY_INCONSISTENT, MerkleTree.differenceHelper(ltree, rtree, new ArrayList<>(), new MerkleTree.TreeDifference(ltree.fullRange.left, ltree.fullRange.right, (byte) 0))); } /** @@@ -530,7 -530,7 +534,7 @@@ { // consume the stack hash = Hashable.binaryHash(hstack.pop(), hash); -- depth = dstack.pop()-1; ++ depth = dstack.pop() - 1; } dstack.push(depth); hstack.push(hash); @@@ -563,4 -563,4 +567,80 @@@ return endOfData(); } } ++ ++ @Test ++ public void testEstimatedSizes() ++ { ++ // With no or negative allowed space we should still get a depth of 1 ++ Assert.assertEquals(1, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, -20, 32)); ++ Assert.assertEquals(1, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, 0, 32)); ++ Assert.assertEquals(1, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, 1, 32)); ++ ++ // The minimum of 1 megabyte split between RF=3 should yield trees of around 10 ++ Assert.assertEquals(10, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, ++ 1048576 / 3, 32)); ++ ++ // With a single megabyte of space we should get 12 ++ Assert.assertEquals(12, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, ++ 1048576, 32)); ++ ++ // With 100 megabytes we should get a limit of 19 ++ Assert.assertEquals(19, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, ++ 100 * 1048576, 32)); ++ ++ // With 300 megabytes we should get the old limit of 20 ++ Assert.assertEquals(20, MerkleTree.estimatedMaxDepthForBytes(Murmur3Partitioner.instance, ++ 300 * 1048576, 32)); ++ Assert.assertEquals(20, MerkleTree.estimatedMaxDepthForBytes(RandomPartitioner.instance, ++ 300 * 1048576, 32)); ++ Assert.assertEquals(20, MerkleTree.estimatedMaxDepthForBytes(ByteOrderedPartitioner.instance, ++ 300 * 1048576, 32)); ++ } ++ ++ @Test ++ public void testEstimatedSizesRealMeasurement() ++ { ++ // Use a fixed source of randomness so that the test does not flake. ++ Random random = new Random(1); ++ checkEstimatedSizes(RandomPartitioner.instance, random); ++ checkEstimatedSizes(Murmur3Partitioner.instance, random); ++ } ++ ++ private void checkEstimatedSizes(IPartitioner partitioner, Random random) ++ { ++ Range<Token> fullRange = new Range<>(partitioner.getMinimumToken(), partitioner.getMinimumToken()); ++ MerkleTree tree = new MerkleTree(partitioner, fullRange, RECOMMENDED_DEPTH, 0); ++ ++ // Test 16 kilobyte -> 16 megabytes ++ for (int i = 14; i < 24; i ++) ++ { ++ long numBytes = 1 << i; ++ int maxDepth = MerkleTree.estimatedMaxDepthForBytes(partitioner, numBytes, 32); ++ long realSizeOfMerkleTree = measureTree(tree, fullRange, maxDepth, random); ++ long biggerTreeSize = measureTree(tree, fullRange, maxDepth + 1, random); ++ ++ Assert.assertTrue(realSizeOfMerkleTree < numBytes); ++ Assert.assertTrue(biggerTreeSize > numBytes); ++ } ++ } ++ ++ private long measureTree(MerkleTree tree, Range<Token> fullRange, int depth, Random random) ++ { ++ tree = new MerkleTree(tree.partitioner(), fullRange, RECOMMENDED_DEPTH, (long) Math.pow(2, depth)); ++ // Initializes it as a fully balanced tree. ++ tree.init(); ++ ++ byte[] key = new byte[128]; ++ // Try to actually allocate some hashes. Note that this is not guaranteed to actually populate the tree, ++ // but we re-use the source of randomness to try to make it reproducible. ++ for (int i = 0; i < tree.maxsize() * 8; i++) ++ { ++ random.nextBytes(key); ++ Token token = tree.partitioner().getToken(ByteBuffer.wrap(key)); ++ tree.get(token).addHash(new RowHash(token, new byte[32], 32)); ++ } ++ ++ tree.hash(fullRange); ++ return ObjectSizes.measureDeep(tree); ++ } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org