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

Forced incremental repairs should promote sstables if they can 

Patch by Blake Eggleston; Reviewed by Marcus Eriksson for CASSANDRA-14294
This commit is contained in:
Blake Eggleston 2018-03-07 09:35:02 -08:00
parent de12f29dc6
commit d4dfbb5c67
6 changed files with 46 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0
* Forced incremental repairs should promote sstables if they can (CASSANDRA-14294)
* Use Murmur3 for validation compactions (CASSANDRA-14002)
* Comma at the end of the seed list is interpretated as localhost (CASSANDRA-14285)
* Refactor read executor and response resolver, abstract read repair (CASSANDRA-14058)

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

@ -234,7 +234,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
traceState = null;
}
final Set<InetAddressAndPort> allNeighbors = new HashSet<>();
Set<InetAddressAndPort> allNeighbors = new HashSet<>();
List<CommonRange> commonRanges = new ArrayList<>();
//pre-calculate output of getLocalRanges and pass it to getNeighbors to increase performance and prevent
@ -286,9 +286,18 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
SystemDistributedKeyspace.startParentRepair(parentSession, keyspace, cfnames, options);
}
boolean force = options.isForcedRepair();
if (force && options.isIncremental())
{
Set<InetAddressAndPort> actualNeighbors = Sets.newHashSet(Iterables.filter(allNeighbors, FailureDetector.instance::isAlive));
force = !allNeighbors.equals(actualNeighbors);
allNeighbors = actualNeighbors;
}
try (Timer.Context ctx = Keyspace.open(keyspace).metric.repairPrepareTime.time())
{
ActiveRepairService.instance.prepareForRepair(parentSession, FBUtilities.getBroadcastAddressAndPort(), allNeighbors, options, columnFamilyStores);
ActiveRepairService.instance.prepareForRepair(parentSession, FBUtilities.getBroadcastAddressAndPort(), allNeighbors, options, force, columnFamilyStores);
progress.incrementAndGet();
}
catch (Throwable t)
@ -307,7 +316,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
else if (options.isIncremental())
{
incrementalRepair(parentSession, startTime, options.isForcedRepair(), traceState, allNeighbors, commonRanges, cfnames);
incrementalRepair(parentSession, startTime, force, traceState, allNeighbors, commonRanges, cfnames);
}
else
{
@ -398,15 +407,14 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
String... cfnames)
{
// the local node also needs to be included in the set of participants, since coordinator sessions aren't persisted
Predicate<InetAddressAndPort> isAlive = FailureDetector.instance::isAlive;
Set<InetAddressAndPort> allParticipants = ImmutableSet.<InetAddressAndPort>builder()
.addAll(forceRepair ? Iterables.filter(allNeighbors, isAlive) : allNeighbors)
.addAll(allNeighbors)
.add(FBUtilities.getBroadcastAddressAndPort())
.build();
List<CommonRange> allRanges = filterCommonRanges(commonRanges, allParticipants, forceRepair);
CoordinatorSession coordinatorSession = ActiveRepairService.instance.consistent.coordinated.registerSession(parentSession, allParticipants);
CoordinatorSession coordinatorSession = ActiveRepairService.instance.consistent.coordinated.registerSession(parentSession, allParticipants, forceRepair);
ListeningExecutorService executor = createExecutor();
AtomicBoolean hasFailure = new AtomicBoolean(false);
ListenableFuture repairResult = coordinatorSession.execute(() -> submitRepairSessions(parentSession, true, executor, allRanges, cfnames),

9
src/java/org/apache/cassandra/repair/consistent/CoordinatorSessions.java
View File

@ -43,10 +43,15 @@ public class CoordinatorSessions
return new CoordinatorSession(builder);
}
public synchronized CoordinatorSession registerSession(UUID sessionId, Set<InetAddressAndPort> participants)
public synchronized CoordinatorSession registerSession(UUID sessionId, Set<InetAddressAndPort> participants, boolean isForced)
{
Preconditions.checkArgument(!sessions.containsKey(sessionId), "A coordinator already exists for session %s", sessionId);
ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(sessionId);
Preconditions.checkArgument(!sessions.containsKey(sessionId),
"A coordinator already exists for session %s", sessionId);
Preconditions.checkArgument(!isForced || prs.repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE,
"cannot promote data for forced incremental repairs");
CoordinatorSession.Builder builder = CoordinatorSession.builder();
builder.withState(ConsistentSession.State.PREPARING);
builder.withSessionID(sessionId);

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

@ -379,11 +379,12 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
* incremental repairs, forced incremental repairs, and full repairs, the UNREPAIRED_SSTABLE value will prevent
* sstables from being promoted to repaired or preserve the repairedAt/pendingRepair values, respectively.
*/
static long getRepairedAt(RepairOption options)
static long getRepairedAt(RepairOption options, boolean force)
{
// we only want to set repairedAt for incremental repairs including all replicas for a token range. For non-global incremental repairs, forced incremental repairs, and
// full repairs, the UNREPAIRED_SSTABLE value will prevent sstables from being promoted to repaired or preserve the repairedAt/pendingRepair values, respectively.
if (options.isIncremental() && options.isGlobal() && !options.isForcedRepair())
// we only want to set repairedAt for incremental repairs including all replicas for a token range. For non-global incremental repairs, full repairs, the UNREPAIRED_SSTABLE value will prevent
// sstables from being promoted to repaired or preserve the repairedAt/pendingRepair values, respectively. For forced repairs, repairedAt time is only set to UNREPAIRED_SSTABLE if we actually
// end up skipping replicas
if (options.isIncremental() && options.isGlobal() && ! force)
{
return Clock.instance.currentTimeMillis();
}
@ -393,9 +394,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
}
}
public UUID prepareForRepair(UUID parentRepairSession, InetAddressAndPort coordinator, Set<InetAddressAndPort> endpoints, RepairOption options, List<ColumnFamilyStore> columnFamilyStores)
public UUID prepareForRepair(UUID parentRepairSession, InetAddressAndPort coordinator, Set<InetAddressAndPort> endpoints, RepairOption options, boolean isForcedRepair, List<ColumnFamilyStore> columnFamilyStores)
{
long repairedAt = getRepairedAt(options);
long repairedAt = getRepairedAt(options, isForcedRepair);
registerParentRepairSession(parentRepairSession, coordinator, columnFamilyStores, options.getRanges(), options.isIncremental(), repairedAt, options.isGlobal(), options.getPreviewKind());
final CountDownLatch prepareLatch = new CountDownLatch(endpoints.size());
final AtomicBoolean status = new AtomicBoolean(true);
@ -434,7 +435,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
}
else
{
if (options.isForcedRepair())
// we pre-filter the endpoints we want to repair for forced incremental repairs. So if any of the
// remaining ones go down, we still want to fail so we don't create repair sessions that can't complete
if (isForcedRepair && !options.isIncremental())
{
prepareLatch.countDown();
}

12
test/unit/org/apache/cassandra/repair/consistent/CoordinatorSessionsTest.java
View File

@ -93,9 +93,9 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
return (InstrumentedCoordinatorSession) super.getSession(sessionId);
}
public InstrumentedCoordinatorSession registerSession(UUID sessionId, Set<InetAddressAndPort> peers)
public InstrumentedCoordinatorSession registerSession(UUID sessionId, Set<InetAddressAndPort> peers, boolean isForced)
{
return (InstrumentedCoordinatorSession) super.registerSession(sessionId, peers);
return (InstrumentedCoordinatorSession) super.registerSession(sessionId, peers, isForced);
}
}
@ -118,7 +118,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
{
CoordinatorSessions sessions = new CoordinatorSessions();
UUID sessionID = registerSession();
CoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS);
CoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false);
Assert.assertEquals(ConsistentSession.State.PREPARING, session.getState());
Assert.assertEquals(sessionID, session.sessionID);
@ -139,7 +139,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
InstrumentedCoordinatorSessions sessions = new InstrumentedCoordinatorSessions();
UUID sessionID = registerSession();
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS);
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false);
Assert.assertEquals(0, session.prepareResponseCalls);
sessions.handlePrepareResponse(new PrepareConsistentResponse(sessionID, PARTICIPANT1, true));
@ -164,7 +164,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
InstrumentedCoordinatorSessions sessions = new InstrumentedCoordinatorSessions();
UUID sessionID = registerSession();
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS);
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false);
Assert.assertEquals(0, session.finalizePromiseCalls);
sessions.handleFinalizePromiseMessage(new FinalizePromise(sessionID, PARTICIPANT1, true));
@ -189,7 +189,7 @@ public class CoordinatorSessionsTest extends AbstractRepairTest
InstrumentedCoordinatorSessions sessions = new InstrumentedCoordinatorSessions();
UUID sessionID = registerSession();
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS);
InstrumentedCoordinatorSession session = sessions.registerSession(sessionID, PARTICIPANTS, false);
Assert.assertEquals(0, session.failCalls);
sessions.handleFailSessionMessage(new FailSession(sessionID));

14
test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
View File

@ -336,22 +336,24 @@ public class ActiveRepairServiceTest
public void repairedAt() throws Exception
{
// regular incremental repair
Assert.assertNotEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true))));
Assert.assertNotEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true)), false));
// subrange incremental repair
Assert.assertNotEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true),
RANGES_KEY, "1:2")));
RANGES_KEY, "1:2"), false));
// hosts incremental repair
Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true),
HOSTS_KEY, "127.0.0.1")));
HOSTS_KEY, "127.0.0.1"), false));
// dc incremental repair
Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true),
DATACENTERS_KEY, "DC2")));
DATACENTERS_KEY, "DC2"), false));
// forced incremental repair
Assert.assertNotEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true),
FORCE_REPAIR_KEY, b2s(true)), false));
Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(true),
FORCE_REPAIR_KEY, b2s(true))));
FORCE_REPAIR_KEY, b2s(true)), true));
// full repair
Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(false))));
Assert.assertEquals(UNREPAIRED_SSTABLE, getRepairedAt(opts(INCREMENTAL_KEY, b2s(false)), false));
}
}