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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Blake Eggleston 2019-02-14 13:19:57 -08:00
parent 4013eacd9d b30c8c98a5
commit 126853068d
16 changed files with 510 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,5 +1,6 @@
3.11.5
Merged from 3.0:
* Improve merkle tree size and time on heap (CASSANDRA-14096)
* Add missing commands to nodetool_completion (CASSANDRA-14916)
* Anti-compaction temporarily corrupts sstable state for readers (CASSANDRA-15004)
Merged from 2.2:

4
NEWS.txt
View File

@ -47,8 +47,8 @@ using the provided 'sstableupgrade' tool.
Upgrading
---------
- Nothing specific to this release, but please see previous upgrading sections,
especially if you are upgrading from 2.2.
- 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.
3.11.4
======

13
conf/cassandra.yaml
View File

@ -498,6 +498,19 @@ concurrent_materialized_view_writes: 32
# 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.
#
# For more details see https://issues.apache.org/jira/browse/CASSANDRA-14096.
#
# repair_session_max_tree_depth: 18
# Total space to use for commit logs on disk.
#
# If space gets above this value, Cassandra will flush every dirty CF

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

@ -123,6 +123,9 @@ public class Config
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;
public int storage_port = 7000;
public int ssl_storage_port = 7001;
public String listen_address;

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

@ -437,6 +437,11 @@ public class DatabaseDescriptor
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.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);
@ -2280,6 +2285,22 @@ public class DatabaseDescriptor
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 getIndexSummaryResizeIntervalInMinutes()
{
return conf.index_summary_resize_interval_in_minutes;

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

@ -1461,8 +1461,12 @@ public class CompactionManager implements CompactionManagerMBean
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 configurable depth (default 18) to prevent large tree (CASSANDRA-11390, CASSANDRA-14096)
int maxDepth = rangeOwningRatio > 0
? (int) Math.floor(Math.max(0.0, DatabaseDescriptor.getRepairSessionMaxTreeDepth() -
Math.log(1 / rangeOwningRatio) / Math.log(2)))
: 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);

8
src/java/org/apache/cassandra/repair/LocalSyncTask.java
View File

@ -49,9 +49,9 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler
private final boolean pullRepair;
public LocalSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, long repairedAt, boolean pullRepair)
public LocalSyncTask(RepairJobDesc desc, InetAddress firstEndpoint, InetAddress secondEndpoint, List<Range<Token>> rangesToSync, long repairedAt, boolean pullRepair)
{
super(desc, r1, r2);
super(desc, firstEndpoint, secondEndpoint, rangesToSync);
this.repairedAt = repairedAt;
this.pullRepair = pullRepair;
}
@ -64,7 +64,7 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler
{
InetAddress local = FBUtilities.getBroadcastAddress();
// We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
InetAddress dst = secondEndpoint.equals(local) ? firstEndpoint : secondEndpoint;
InetAddress preferred = SystemKeyspace.getPreferredIP(dst);
String message = String.format("Performing streaming repair of %d ranges with %s", differences.size(), dst);
@ -117,7 +117,7 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler
public void onSuccess(StreamState result)
{
String message = String.format("Sync complete using session %s between %s and %s on %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
String message = String.format("Sync complete using session %s between %s and %s on %s", desc.sessionId, firstEndpoint, secondEndpoint, desc.columnFamily);
logger.info("[repair #{}] {}", desc.sessionId, message);
Tracing.traceRepair(message);
set(stat);

8
src/java/org/apache/cassandra/repair/RemoteSyncTask.java
View File

@ -41,15 +41,15 @@ public class RemoteSyncTask extends SyncTask
{
private static final Logger logger = LoggerFactory.getLogger(RemoteSyncTask.class);
public RemoteSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2)
public RemoteSyncTask(RepairJobDesc desc, InetAddress firstEndpoint, InetAddress secondEndpoint, List<Range<Token>> rangesToSync)
{
super(desc, r1, r2);
super(desc, firstEndpoint, secondEndpoint, rangesToSync);
}
protected void startSync(List<Range<Token>> differences)
{
InetAddress local = FBUtilities.getBroadcastAddress();
SyncRequest request = new SyncRequest(desc, local, r1.endpoint, r2.endpoint, differences);
SyncRequest request = new SyncRequest(desc, local, firstEndpoint, secondEndpoint, differences);
String message = String.format("Forwarding streaming repair of %d ranges to %s (to be streamed with %s)", request.ranges.size(), request.src, request.dst);
logger.info("[repair #{}] {}", desc.sessionId, message);
Tracing.traceRepair(message);
@ -64,7 +64,7 @@ public class RemoteSyncTask extends SyncTask
}
else
{
setException(new RepairException(desc, String.format("Sync failed between %s and %s", r1.endpoint, r2.endpoint)));
setException(new RepairException(desc, String.format("Sync failed between %s and %s", firstEndpoint, secondEndpoint)));
}
}
}

65
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -20,14 +20,18 @@ package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.*;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.Pair;
/**
@ -105,33 +109,7 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
{
public ListenableFuture<List<SyncStat>> apply(List<TreeResponse> trees)
{
InetAddress local = FBUtilities.getLocalAddress();
List<SyncTask> syncTasks = new ArrayList<>();
// We need to difference all trees one against another
for (int i = 0; i < trees.size() - 1; ++i)
{
TreeResponse r1 = trees.get(i);
for (int j = i + 1; j < trees.size(); ++j)
{
TreeResponse r2 = trees.get(j);
SyncTask task;
if (r1.endpoint.equals(local) || r2.endpoint.equals(local))
{
task = new LocalSyncTask(desc, r1, r2, repairedAt, session.pullRepair);
}
else
{
task = new RemoteSyncTask(desc, r1, r2);
// RemoteSyncTask expects SyncComplete message sent back.
// Register task to RepairSession to receive response.
session.waitForSync(Pair.create(desc, new NodePair(r1.endpoint, r2.endpoint)), (RemoteSyncTask) task);
}
syncTasks.add(task);
taskExecutor.submit(task);
}
}
return Futures.allAsList(syncTasks);
return Futures.allAsList(createSyncTasks(trees, FBUtilities.getLocalAddress()));
}
}, taskExecutor);
@ -160,6 +138,39 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
Futures.getUnchecked(validations);
}
@VisibleForTesting
List<SyncTask> createSyncTasks(List<TreeResponse> trees, InetAddress local)
{
List<SyncTask> syncTasks = new ArrayList<>();
// We need to difference all trees one against another
for (int i = 0; i < trees.size() - 1; ++i)
{
TreeResponse r1 = trees.get(i);
for (int j = i + 1; j < trees.size(); ++j)
{
TreeResponse r2 = trees.get(j);
SyncTask task;
List<Range<Token>> differences = MerkleTrees.difference(r1.trees, r2.trees);
if (r1.endpoint.equals(local) || r2.endpoint.equals(local))
{
task = new LocalSyncTask(desc, r1.endpoint, r2.endpoint, differences, repairedAt, session.pullRepair);
}
else
{
task = new RemoteSyncTask(desc, r1.endpoint, r2.endpoint, differences);
// RemoteSyncTask expects SyncComplete message sent back.
// Register task to RepairSession to receive response.
session.waitForSync(Pair.create(desc, new NodePair(r1.endpoint, r2.endpoint)), (RemoteSyncTask) task);
}
syncTasks.add(task);
taskExecutor.submit(task);
}
}
return syncTasks;
}
/**
* Creates {@link ValidationTask} and submit them to task executor in parallel.
*

20
src/java/org/apache/cassandra/repair/RepairSession.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
@ -53,8 +54,8 @@ import org.apache.cassandra.utils.Pair;
* validationComplete()).
* </li>
* <li>Synchronization phase: once all trees are received, the job compares each tree with
* all the other using a so-called {@link SyncTask}. If there is difference between 2 trees, the
* concerned SyncTask will start a streaming of the difference between the 2 endpoint concerned.
* all the others and creates a {@link SyncTask} for each diverging replica. If there are differences
* between 2 trees, the concerned SyncTask stream the differences between the 2 endpoints concerned.
* </li>
* </ol>
* The job is done once all its SyncTasks are done (i.e. have either computed no differences
@ -100,7 +101,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
private final ConcurrentMap<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> 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;
private volatile boolean terminated = false;
@ -138,6 +139,13 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
this.endpoints = endpoints;
this.repairedAt = repairedAt;
this.pullRepair = pullRepair;
this.taskExecutor = MoreExecutors.listeningDecorator(createExecutor());
}
@VisibleForTesting
protected DebuggableThreadPoolExecutor createExecutor()
{
return DebuggableThreadPoolExecutor.createCachedThreadpoolWithMaxSize("RepairJobTask");
}
public UUID getId()
@ -202,6 +210,12 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
task.syncComplete(success);
}
@VisibleForTesting
Map<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> getSyncingTasks()
{
return Collections.unmodifiableMap(syncingTasks);
}
private String repairedNodes()
{
StringBuilder sb = new StringBuilder();

34
src/java/org/apache/cassandra/repair/SyncTask.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.List;
import com.google.common.util.concurrent.AbstractFuture;
@ -26,10 +27,9 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.MerkleTrees;
/**
* SyncTask will calculate the difference of MerkleTree between two nodes
* SyncTask takes the difference of MerkleTrees between two nodes
* and perform necessary operation to repair replica.
*/
public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runnable
@ -37,16 +37,19 @@ public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runna
private static Logger logger = LoggerFactory.getLogger(SyncTask.class);
protected final RepairJobDesc desc;
protected final TreeResponse r1;
protected final TreeResponse r2;
protected final InetAddress firstEndpoint;
protected final InetAddress secondEndpoint;
private final List<Range<Token>> rangesToSync;
protected volatile SyncStat stat;
public SyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2)
public SyncTask(RepairJobDesc desc, InetAddress firstEndpoint, InetAddress secondEndpoint, List<Range<Token>> rangesToSync)
{
this.desc = desc;
this.r1 = r1;
this.r2 = r2;
this.firstEndpoint = firstEndpoint;
this.secondEndpoint = secondEndpoint;
this.rangesToSync = rangesToSync;
}
/**
@ -54,25 +57,22 @@ public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runna
*/
public void run()
{
// compare trees, and collect differences
List<Range<Token>> differences = MerkleTrees.difference(r1.trees, r2.trees);
stat = new SyncStat(new NodePair(r1.endpoint, r2.endpoint), differences.size());
stat = new SyncStat(new NodePair(firstEndpoint, secondEndpoint), rangesToSync.size());
// choose a repair method based on the significance of the difference
String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
if (differences.isEmpty())
String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, firstEndpoint, secondEndpoint, desc.columnFamily);
if (rangesToSync.isEmpty())
{
logger.info(String.format(format, "are consistent"));
Tracing.traceRepair("Endpoint {} is consistent with {} for {}", r1.endpoint, r2.endpoint, desc.columnFamily);
Tracing.traceRepair("Endpoint {} is consistent with {} for {}", firstEndpoint, secondEndpoint, desc.columnFamily);
set(stat);
return;
}
// non-0 difference: perform streaming repair
logger.info(String.format(format, "have " + differences.size() + " range(s) out of sync"));
Tracing.traceRepair("Endpoint {} has {} range(s) out of sync with {} for {}", r1.endpoint, differences.size(), r2.endpoint, desc.columnFamily);
startSync(differences);
logger.info(String.format(format, "have " + rangesToSync.size() + " range(s) out of sync"));
Tracing.traceRepair("Endpoint {} has {} range(s) out of sync with {} for {}", firstEndpoint, rangesToSync.size(), secondEndpoint, desc.columnFamily);
startSync(rangesToSync);
}
public SyncStat getCurrentStat()

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

@ -3688,6 +3688,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
ActiveRepairService.instance.terminateSessions();
}
public void setRepairSessionMaxTreeDepth(int depth)
{
DatabaseDescriptor.setRepairSessionMaxTreeDepth(depth);
}
public int getRepairSessionMaxTreeDepth()
{
return DatabaseDescriptor.getRepairSessionMaxTreeDepth();
}
/* End of MBean interface methods */
/**

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

@ -381,6 +381,10 @@ public interface StorageServiceMBean extends NotificationEmitter
public void forceTerminateAllRepairSessions();
public void setRepairSessionMaxTreeDepth(int depth);
public int getRepairSessionMaxTreeDepth();
/**
* transfer this node's data to other machines and remove it from service.
*/

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

@ -27,6 +27,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@ -44,6 +45,7 @@ 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;
@ -275,15 +277,48 @@ public class DatabaseDescriptorTest
DatabaseDescriptor.applyAddressConfig(testConfig);
}
@Test
public void testTokensFromString()
{
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 testRepairSessionSizeToggles()
{
int previousDepth = DatabaseDescriptor.getRepairSessionMaxTreeDepth();
try
{
Assert.assertEquals(18, 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);
}
}
}

8
test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java
View File

@ -76,7 +76,9 @@ public class LocalSyncTaskTest extends SchemaLoader
// note: we reuse the same endpoint which is bogus in theory but fine here
TreeResponse r1 = new TreeResponse(ep1, tree1);
TreeResponse r2 = new TreeResponse(ep2, tree2);
LocalSyncTask task = new LocalSyncTask(desc, r1, r2, ActiveRepairService.UNREPAIRED_SSTABLE, false);
LocalSyncTask task = new LocalSyncTask(desc, r1.endpoint, r2.endpoint,
MerkleTrees.difference(r1.trees, r2.trees),
ActiveRepairService.UNREPAIRED_SSTABLE, false);
task.run();
assertEquals(0, task.get().numberOfDifferences);
@ -111,7 +113,9 @@ public class LocalSyncTaskTest extends SchemaLoader
// note: we reuse the same endpoint which is bogus in theory but fine here
TreeResponse r1 = new TreeResponse(InetAddress.getByName("127.0.0.1"), tree1);
TreeResponse r2 = new TreeResponse(InetAddress.getByName("127.0.0.2"), tree2);
LocalSyncTask task = new LocalSyncTask(desc, r1, r2, ActiveRepairService.UNREPAIRED_SSTABLE, false);
LocalSyncTask task = new LocalSyncTask(desc, r1.endpoint, r2.endpoint,
MerkleTrees.difference(r1.trees, r2.trees),
ActiveRepairService.UNREPAIRED_SSTABLE, false);
task.run();
// ensure that the changed range was recorded

326
test/unit/org/apache/cassandra/repair/RepairJobTest.java Normal file
View File

@ -0,0 +1,326 @@
/*
* 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.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.TimeUnit;
import java.util.stream.Collectors;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
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.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
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.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.assertTrue;
public class RepairJobTest extends SchemaLoader
{
private static final long TEST_TIMEOUT_S = 10;
private static final long THREAD_TIMEOUT_MILLIS = 100;
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 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;
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)
{
super(parentRepairSession, id, ranges, keyspace, parallelismDegree, endpoints, repairedAt, pullRepair, cfnames);
}
protected DebuggableThreadPoolExecutor createExecutor()
{
DebuggableThreadPoolExecutor executor = super.createExecutor();
executor.setKeepAliveTime(THREAD_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
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");
}
@Before
public void setup()
{
Set<InetAddress> neighbors = new HashSet<>(Arrays.asList(addr2, addr3));
UUID parentRepairSession = UUID.randomUUID();
ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, FBUtilities.getBroadcastAddress(),
Collections.singletonList(Keyspace.open(KEYSPACE).getColumnFamilyStore(CF)), fullRange, false,
ActiveRepairService.UNREPAIRED_SSTABLE, false);
this.session = new MeasureableRepairSession(parentRepairSession, UUIDGen.getTimeUUID(), fullRange,
KEYSPACE, RepairParallelism.SEQUENTIAL, neighbors,
ActiveRepairService.UNREPAIRED_SSTABLE, false, CF);
this.job = new RepairJob(session, CF);
this.sessionJobDesc = new RepairJobDesc(session.parentRepairSession, session.getId(),
session.keyspace, CF, session.getRanges());
DatabaseDescriptor.setBroadcastAddress(addr1);
}
@After
public void reset()
{
ActiveRepairService.instance.terminateSessions();
MessagingService.instance().clearMessageSinks();
}
/**
* Ensure we can do an end to end repair of consistent data and get the messages we expect
*/
@Test
public void testEndToEndNoDifferences() throws Exception
{
Map<InetAddress, MerkleTrees> mockTrees = new HashMap<>();
mockTrees.put(FBUtilities.getBroadcastAddress(), 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);
// 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));
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());
long singleTreeSize = ObjectSizes.measureDeep(mockTrees.get(addr2));
// 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);
// SyncTasks themselves should not contain significant memory
assertTrue(ObjectSizes.measureDeep(syncTasks) < 0.8 * 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);
// The session can retain 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)
{
// The measured size of the syncingTasks, and result of the computation should be much smaller
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);
assertEquals(3, results.size());
// Should be two RemoteSyncTasks with ranges and one empty one
assertExpectedDifferences(new ArrayList<>(session.getSyncingTasks().values()), 1, 1, 0);
int numDifferent = 0;
for (SyncStat stat : results)
{
if (stat.nodes.endpoint1.equals(addr2) || stat.nodes.endpoint2.equals(addr2))
{
assertEquals(1, stat.numberOfDifferences);
numDifferent++;
}
}
assertEquals(2, numDifferent);
}
private void assertExpectedDifferences(Collection<RemoteSyncTask> tasks, Integer ... differences)
{
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));
}
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,
List<MessageOut> messageCapture)
{
MessagingService.instance().addMessageSink(new IMessageSink()
{
public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress 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);
break;
default:
break;
}
return false;
}
public boolean allowIncomingMessage(MessageIn message, int id)
{
return message.verb == MessagingService.Verb.REQUEST_RESPONSE;
}
});
}
}