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.1' into trunk

This commit is contained in:
Marcus Eriksson 2015-11-17 10:15:58 +01:00
parent 0010fce6d2 798a7a8fe8
commit 6645062caa
5 changed files with 41 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -13,10 +13,12 @@ Merged from 3.0:
* Keep the file open in trySkipCache (CASSANDRA-10669)
* Updated trigger example (CASSANDRA-10257)
Merged from 2.2:
* Don't do anticompaction after subrange repair (CASSANDRA-10422)
* Fix SimpleDateType type compatibility (CASSANDRA-10027)
* (Hadoop) fix splits calculation (CASSANDRA-10640)
* (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058)
Merged from 2.1:
* Reject incremental repair with subrange repair (CASSANDRA-10422)
* Add a nodetool command to refresh size_estimates (CASSANDRA-9579)
* Invalidate cache after stream receive task is completed (CASSANDRA-10341)
* Reject counter writes in CQLSSTableWriter (CASSANDRA-10258)

18
src/java/org/apache/cassandra/repair/messages/RepairOption.java
View File

@ -144,6 +144,10 @@ public class RepairOption
Set<Range<Token>> ranges = new HashSet<>();
if (rangesStr != null)
{
if (incremental)
logger.warn("Incremental repair can't be requested with subrange repair " +
"because each subrange repair would generate an anti-compacted table. " +
"The repair will occur but without anti-compaction.");
StringTokenizer tokenizer = new StringTokenizer(rangesStr, ",");
while (tokenizer.hasMoreTokens())
{
@ -158,7 +162,7 @@ public class RepairOption
}
}
RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges);
RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, !ranges.isEmpty());
// data centers
String dataCentersStr = options.get(DATACENTERS_KEY);
@ -217,13 +221,14 @@ public class RepairOption
private final boolean incremental;
private final boolean trace;
private final int jobThreads;
private final boolean isSubrangeRepair;
private final Collection<String> columnFamilies = new HashSet<>();
private final Collection<String> dataCenters = new HashSet<>();
private final Collection<String> hosts = new HashSet<>();
private final Collection<Range<Token>> ranges = new HashSet<>();
public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges)
public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean isSubrangeRepair)
{
if (FBUtilities.isWindows() &&
(DatabaseDescriptor.getDiskAccessMode() != Config.DiskAccessMode.standard || DatabaseDescriptor.getIndexAccessMode() != Config.DiskAccessMode.standard) &&
@ -240,6 +245,7 @@ public class RepairOption
this.trace = trace;
this.jobThreads = jobThreads;
this.ranges.addAll(ranges);
this.isSubrangeRepair = isSubrangeRepair;
}
public RepairParallelism getParallelism()
@ -289,8 +295,14 @@ public class RepairOption
public boolean isGlobal()
{
return dataCenters.isEmpty() && hosts.isEmpty();
return dataCenters.isEmpty() && hosts.isEmpty() && !isSubrangeRepair();
}
public boolean isSubrangeRepair()
{
return isSubrangeRepair;
}
@Override
public String toString()
{

2
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -351,6 +351,8 @@ public class ActiveRepairService
{
assert parentRepairSession != null;
ParentRepairSession prs = getParentRepairSession(parentRepairSession);
//A repair will be marked as not global if it is a subrange repair to avoid many small anti-compactions
//in addition to other scenarios such as repairs not involving all DCs or hosts
if (!prs.isGlobal)
{
logger.info("Not a global repair, will not do anticompaction");

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

@ -2910,7 +2910,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
parallelism = RepairParallelism.PARALLEL;
}
RepairOption options = new RepairOption(parallelism, primaryRange, !fullRepair, false, 1, Collections.<Range<Token>>emptyList());
RepairOption options = new RepairOption(parallelism, primaryRange, !fullRepair, false, 1, Collections.<Range<Token>>emptyList(), false);
if (dataCenters != null)
{
options.getDataCenters().addAll(dataCenters);
@ -2992,9 +2992,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
logger.warn("Snapshot-based repair is not yet supported on Windows. Reverting to parallel repair.");
parallelism = RepairParallelism.PARALLEL;
}
if (!fullRepair)
logger.warn("Incremental repair can't be requested with subrange repair " +
"because each subrange repair would generate an anti-compacted table. " +
"The repair will occur but without anti-compaction.");
Collection<Range<Token>> repairingRange = createRepairRangeFrom(beginToken, endToken);
RepairOption options = new RepairOption(parallelism, false, !fullRepair, false, 1, repairingRange);
RepairOption options = new RepairOption(parallelism, false, !fullRepair, false, 1, repairingRange, true);
options.getDataCenters().addAll(dataCenters);
if (hosts != null)
{

17
test/unit/org/apache/cassandra/repair/messages/RepairOptionTest.java
View File

@ -25,6 +25,8 @@ import java.util.Set;
import org.junit.Test;
import com.google.common.collect.ImmutableMap;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
@ -61,7 +63,7 @@ public class RepairOptionTest
Map<String, String> options = new HashMap<>();
options.put(RepairOption.PARALLELISM_KEY, "parallel");
options.put(RepairOption.PRIMARY_RANGE_KEY, "false");
options.put(RepairOption.INCREMENTAL_KEY, "true");
options.put(RepairOption.INCREMENTAL_KEY, "false");
options.put(RepairOption.RANGES_KEY, "0:10,11:20,21:30");
options.put(RepairOption.COLUMNFAMILIES_KEY, "cf1,cf2,cf3");
options.put(RepairOption.DATACENTERS_KEY, "dc1,dc2,dc3");
@ -70,7 +72,7 @@ public class RepairOptionTest
option = RepairOption.parse(options, partitioner);
assertTrue(option.getParallelism() == RepairParallelism.PARALLEL);
assertFalse(option.isPrimaryRange());
assertTrue(option.isIncremental());
assertFalse(option.isIncremental());
Set<Range<Token>> expectedRanges = new HashSet<>(3);
expectedRanges.add(new Range<>(tokenFactory.fromString("0"), tokenFactory.fromString("10")));
@ -96,4 +98,15 @@ public class RepairOptionTest
expectedHosts.add("127.0.0.3");
assertEquals(expectedHosts, option.getHosts());
}
@Test
public void testIncrementalRepairWithSubrangesIsNotGlobal() throws Exception
{
RepairOption ro = RepairOption.parse(ImmutableMap.of(RepairOption.INCREMENTAL_KEY, "true", RepairOption.RANGES_KEY, "42:42"),
Murmur3Partitioner.instance);
assertFalse(ro.isGlobal());
ro = RepairOption.parse(ImmutableMap.of(RepairOption.INCREMENTAL_KEY, "true", RepairOption.RANGES_KEY, ""),
Murmur3Partitioner.instance);
assertTrue(ro.isGlobal());
}
}